FastMCP 集成 Scalekit OAuth:用 Resource Server 模式保护你的 MCP 服务端
【免费下载链接】fastmcp🚀 The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp
本指南以仓库中的 Scalekit OAuth 示例为核心,完整讲解如何用 Scalekit 的 OAuth 2.1 能力保护 FastMCP 服务端:从 Scalekit 控制台注册 MCP Server、配置环境变量,到编写带鉴权的服务端与自动完成 OAuth 授权的客户端,再到深入ScalekitProvider的源码原理与验证机制。读完本文,你将能够独立复现一个「未登录无法调用工具」的受保护 MCP 服务,并理解其背后的 JWT 校验与元数据转发流程。
示例概览:一条完整的 Scalekit OAuth 链路
示例位于仓库的 examples/auth/scalekit_oauth/ 目录,包含三个文件:
| 文件 | 作用 |
|---|---|
| README.md | 官方示例说明,包含配置与运行步骤 |
| server.py | 受 Scalekit OAuth 保护的 FastMCP 服务端 |
| client.py | 自动完成 OAuth 授权并调用受保护工具的客户端 |
它演示的是 MCP 生态中典型的Remote OAuth / Resource Server 模式:Scalekit 负责用户认证与签发访问令牌,FastMCP 服务端作为受保护的资源服务器,只接受携带合法访问令牌的请求。整个示例跑通后,未认证的客户端将无法列出或调用任何工具,而使用 OAuth 完成登录的客户端可以正常使用全部工具。
前置准备:在 Scalekit 控制台注册 MCP Server
在写任何代码之前,需要先在 Scalekit 侧完成资源配置,这一步决定了后续环境变量的取值。
创建 Scalekit 账号并获取凭证
- 前往 Scalekit Dashboard 注册账号;
- 从Developers → Settings复制你的Environment URL(形如
https://your-env.scalekit.com); - 进入Developers → MCP Servers查看Resource ID(形如
res_xxx)。
注册你的 MCP Server
- 进入MCP Servers页面,选择Create New Server;
- 填写 MCP Server 的详细信息(名称、资源标识符以及期望的 MCP 客户端认证设置);
- 保存后复制生成的Resource ID(例如
res_123)。
值得注意的一点:在 Scalekit 中注册资源时,确保Resource Identifier 与你为 FastMCP 配置的 MCP URL 完全一致。从 scalekit.py 的模块文档可以看到,这是官方明确标注的 IMPORTANT SETUP REQUIREMENTS,一旦不一致,后续令牌的 audience 校验就会失败。
配置环境变量:.env 文件的完整说明
在示例目录创建.env文件,填入以下变量:
# Required Scalekit credentials SCALEKIT_ENVIRONMENT_URL=<YOUR_APP_ENVIRONMENT_URL> SCALEKIT_RESOURCE_ID=<YOUR_APP_RESOURCE_ID> # res_926EXAMPLE5878 BASE_URL=http://127.0.0.1:8000/ # Optional: additional scopes tokens must include (comma-separated) # SCALEKIT_REQUIRED_SCOPES=read,write各变量的语义与取值规则如下:
| 变量 | 必填 | 含义 | 说明 |
|---|---|---|---|
SCALEKIT_ENVIRONMENT_URL | 是 | Scalekit 环境 URL | 例如https://your-env.scalekit.com,对应ScalekitProvider.environment_url |
SCALEKIT_RESOURCE_ID | 是 | Scalekit 资源 ID | 上一步在控制台复制的res_xxx,作为 JWT 的 audience 校验依据 |
BASE_URL | 否 | FastMCP 服务对外暴露的地址 | 默认http://127.0.0.1:8000/,开发时可为 localhost |
SCALEKIT_REQUIRED_SCOPES | 否 | 令牌必须携带的 scope(逗号分隔) | 例如read,write;不设置则不强制 scope 校验 |
在 server.py 中可以看到这些变量的读取逻辑:SCALEKIT_REQUIRED_SCOPES会被按逗号切分并去除空白字符,转成list[str]传给 provider;SCALEKIT_ENVIRONMENT_URL与SCALEKIT_RESOURCE_ID若缺失,会分别退化为占位值https://your-env.scalekit.com和空字符串,因此生产环境务必显式提供真实凭证。
注意:仓库中的
.env不会被自动读取。官方的 集成文档 明确提示「Nothing reads.envautomatically」,需要在代码中通过python-dotenv(pip install python-dotenv)显式加载,例如在构造 provider 前调用load_dotenv()。
服务端实现:用 ScalekitProvider 一键开启 OAuth
完整代码
server.py 的完整实现如下:
import os from fastmcp import FastMCP from fastmcp.server.auth.providers.scalekit import ScalekitProvider required_scopes_env = os.getenv("SCALEKIT_REQUIRED_SCOPES") required_scopes = ( [scope.strip() for scope in required_scopes_env.split(",") if scope.strip()] if required_scopes_env else None ) auth = ScalekitProvider( environment_url=os.getenv("SCALEKIT_ENVIRONMENT_URL") or "https://your-env.scalekit.com", resource_id=os.getenv("SCALEKIT_RESOURCE_ID") or "", base_url=os.getenv("BASE_URL", "http://127.0.0.1:8000/"), required_scopes=required_scopes, ) mcp = FastMCP("Scalekit OAuth Example Server", auth=auth) @mcp.tool def echo(message: str) -> str: """Echo the provided message.""" return message @mcp.tool def auth_status() -> dict: """Show Scalekit authentication status.""" # In a real implementation, you would extract user info from the JWT token return { "message": "This tool requires authentication via Scalekit", "authenticated": True, "provider": "Scalekit", } if __name__ == "__main__": mcp.run(transport="http", port=8000)核心只有三步:从环境变量构造ScalekitProvider,把它作为auth参数传给FastMCP(...),然后用 HTTP 传输在 8000 端口启动。之后所有对 MCP 端点的请求都会先经过 OAuth 校验。
ScalekitProvider 的完整参数表
从 scalekit.py 的构造器签名可以整理出全部参数:
| 参数 | 必填 | 类型 | 说明 |
|---|---|---|---|
environment_url | 是 | AnyHttpUrl \| str | Scalekit 环境 URL,如https://your-env.scalekit.com |
resource_id | 是 | str | Scalekit 资源 ID(res_xxx) |
base_url | 否 | AnyHttpUrl \| str | 本 FastMCP 服务的公网地址 |
mcp_url | 否 | AnyHttpUrl \| str | base_url的已废弃别名,未来版本将移除 |
client_id | 否 | str | 已废弃参数,不再需要,仅向后兼容 |
required_scopes | 否 | list[str] \| None | 令牌必须包含的 scope 列表 |
scopes_supported | 否 | list[str] \| None | 在 OAuth 元数据中通告的 scope;为 None 时使用required_scopes,适合「客户端申请的 scope」与「服务端强制的 scope」不一致的场景 |
resource_name | 否 | str \| None | 受保护资源的元数据名称 |
resource_documentation | 否 | AnyHttpUrl \| None | 受保护资源的文档 URL |
token_verifier | 否 | TokenVerifier \| None | 自定义令牌验证器;为 None 时自动创建适配 Scalekit 的JWTVerifier |
两个值得注意的兼容性细节(均有源码佐证):
mcp_url与client_id已废弃:源码在两者被传入时会打印 deprecation warning(scalekit.py),且当base_url与mcp_url同时提供时优先使用base_url。对应测试 test_scalekit.py 验证了这两种行为;- URL 尾斜杠被规范化:
environment_url会rstrip("/"),base_url会以/结尾(scalekit.py),避免拼接端点时出现双斜杠问题,测试 test_scalekit.py 也覆盖了该场景。
默认 JWT 验证器的构造逻辑
当不传token_verifier时,ScalekitProvider会自动构建一个JWTVerifier(scalekit.py):
token_verifier = JWTVerifier( jwks_uri=f"{self.environment_url}/keys", issuer=expected_issuers, algorithm="RS256", audience=self.resource_id, required_scopes=self.required_scopes or None, )也就是说,默认验证规则为:从{environment_url}/keys拉取 JWKS 公钥、强制 RS256 签名算法、校验aud等于resource_id、并按需校验 scope。对应单元测试 test_scalekit.py 逐一断言了这些端点与取值。
一个值得了解的实现细节:expected_issuers同时接受裸环境 URL(https://your-env.scalekit.com)和资源级 issuer(https://your-env.scalekit.com/resources/{resource_id})两种形式(scalekit.py)。这是因为 Scalekit 正在将iss声明从裸环境 URL 迁移到资源级 issuer,同时接受两种形式可以确保迁移前后签发的令牌都能通过校验——test_scalekit.py 中的TestScalekitIssuerMigration专门用新旧两种 issuer 构造令牌,验证迁移前、迁移后均通过,而未知 issuer 被拒绝。
运行服务端
在示例目录下启动:
# From this directory uv run python server.py服务启动后监听http://127.0.0.1:8000/mcp,并启用 Scalekit OAuth 认证。此时直接使用普通客户端访问会被拒绝:集成测试 test_scalekit.py 验证了「无凭据的客户端调用list_tools会抛出MCPError(服务端返回 401)」。
在本地开发阶段,BASE_URL使用http://127.0.0.1:8000/即可;生产环境应替换为真实公网地址并使用 HTTPS,官方集成文档对此有明确建议(docs/integrations/scalekit.mdx)。
客户端实现:自动完成 OAuth 授权
client.py 演示了客户端侧的完整流程:
import asyncio from fastmcp.client import Client SERVER_URL = "http://127.0.0.1:8000/mcp" async def main(): try: async with Client(SERVER_URL, auth="oauth") as client: assert await client.ping() print("✅ Successfully authenticated with Scalekit!") tools = await client.list_tools() print(f"🔧 Available tools ({len(tools)}):") for tool in tools: print(f" - {tool.name}: {tool.description}") # Test calling a tool result = await client.call_tool("echo", {"message": "Hello from Scalekit!"}) print(f"🎯 Echo result: {result}") # Test calling auth status tool auth_status = await client.call_tool("auth_status", {}) print(f"👌 Auth status: {auth_status}") except Exception as e: print(f"❌ Authentication failed: {e}") raise if __name__ == "__main__": asyncio.run(main())关键点在于Client(SERVER_URL, auth="oauth"):只要显式声明auth="oauth",客户端便会在连接阶段自动检测服务端返回的401与 OAuth 元数据,然后走完整个授权流程。
按 README 的运行说明,client.py的行为依次是:
- 尝试连接服务端;
- 检测到需要 OAuth 认证(服务端返回未授权);
- 打开浏览器进入 Scalekit 认证页面,完成用户登录与授权;
- 完成 OAuth 流程并连接服务端(客户端按 OAuth 2.1 + PKCE 规范换取访问令牌,令牌随后随请求发送);
- 演示调用受保护的工具:先
ping确认连通,再列出工具清单,最后依次调用echo与auth_status两个工具。
运行命令:
uv run python client.py源码级原理:令牌验证与元数据转发
令牌校验发生在哪里
访问令牌的校验由JWTVerifier.load_access_token完成(jwt.py),其校验链包括:
- 根据令牌头中的
kid从 JWKS 拉取并缓存公钥(缓存 TTL 为 1 小时,见 jwt.py),并跳过无法解析或与算法不匹配的 JWK(符合 RFC 7517 §5 的容错要求); - 拒绝携带不支持的
crit(critical)JWS 头的令牌(jwt.py); - 校验
exp过期时间、ississuer(支持字符串或列表)、audaudience(支持字符串或列表,两者为列表时取交集); - 从
scope或scp声明中提取 scope(兼容不同 IdP 的写法,见 jwt.py),并做required_scopes子集检查; - 全部通过后返回包含
client_id、scopes、expires_at、subject与原始claims的AccessToken。
授权服务器元数据的转发
ScalekitProvider.get_routes(scalekit.py)在标准受保护资源路由之外,额外注册了一个GET /.well-known/oauth-authorization-server端点。该端点将请求转发到 Scalekit 的元数据地址:
{environment_url}/.well-known/oauth-authorization-server/resources/{resource_id}并把上游的 JSON 响应原样返回给客户端。这样 MCP 客户端无需预先配置 Scalekit 的端点,只要访问服务端就能发现完整的授权服务器元数据,从而支持 OAuth 2.1 下的动态客户端注册(DCR)与 PKCE 流程。集成测试 test_scalekit.py 通过 mock 上游验证了该转发行为:请求/根路径下的元数据端点,返回内容与 mock 的 Scalekit 响应完全一致,且上游请求 URL 正确拼接了资源 ID。
从 JWT 中读取用户上下文
认证成功后,令牌声明(claims)可以被注入到工具中用于获取用户上下文。官方集成文档(docs/integrations/scalekit.mdx)给出的模式是使用get_access_token:
from fastmcp.server.dependencies import get_access_token @mcp.tool def inspect_token() -> dict: """Inspect the current JWT token claims.""" token = get_access_token() if token is None: return {"error": "No token found"} # Claims were already verified by the auth provider. return token.claims由于令牌在到达工具层之前已经由认证提供方完成签名、过期、issuer、audience 与 scope 的校验,token.claims可以直接作为可信的用户上下文使用——这也是示例中auth_status工具标注「真实实现中应从 JWT 提取用户信息」的原因。
生产环境建议与故障排查
生产配置模板
官方集成文档(docs/integrations/scalekit.mdx)建议生产环境直接从环境变量加载配置,避免硬编码:
import os from fastmcp import FastMCP from fastmcp.server.auth.providers.scalekit import ScalekitProvider # Load configuration from environment variables auth = ScalekitProvider( environment_url=os.environ["SCALEKIT_ENVIRONMENT_URL"], resource_id=os.environ["SCALEKIT_RESOURCE_ID"], base_url=os.environ.get("BASE_URL", "https://your-server.com") ) mcp = FastMCP(name="My Scalekit Protected Server", auth=auth) @mcp.tool def protected_action() -> str: """A tool that requires authentication.""" return "Access granted via Scalekit!"scope 的取舍
required_scopes的语义值得仔细把握:设置它意味着令牌必须携带这些 scope 才能通过校验(jwt.py 实现的是子集包含检查);不设置则接受该资源签发的任意令牌。官方建议是「需要令牌携带特定权限时设置,否则留空」;如果客户端申请的 scope 与服务端强制的 scope 不一致,则用scopes_supported单独通告客户端应申请的 scope。
开启调试日志
认证问题排查可以从开启 DEBUG 日志入手:
import logging logging.basicConfig(level=logging.DEBUG)ScalekitProvider在初始化、JWT 验证器构建、元数据转发等环节都打了logger.debug日志(见 scalekit.py 与 jwt.py 中的失败原因记录),包括 issuer 不匹配、audience 不匹配、缺少必需 scope、令牌过期等具体原因,能帮助快速定位是凭证配置问题还是令牌本身的问题。
其他要点
- HTTPS:生产环境必须使用 HTTPS,避免令牌在传输中被截获;
- 令牌过期:JWT 中的
exp校验失败属于正常的令牌轮换噪音,被记录为 INFO 级别而非 WARNING(jwt.py),不必视为异常; - Enterprise SSO:Scalekit 支持 SAML、OIDC、OAuth 2.0、ADFS、Azure AD、Google Workspace 等企业级身份源,配合 OAuth 2.1/DCR,客户端可以无需预置凭证即完成自注册(docs/integrations/scalekit.mdx)。
小结
本示例完整覆盖了「Scalekit 控制台注册 → 环境变量配置 → 受保护服务端编写 → OAuth 客户端连接」的闭环,而这套能力在框架层的落点只有一个类:ScalekitProvider。它承担了 JWT 校验规则构建、授权服务器元数据转发、scope 强制等全部 OAuth 集成工作,服务端代码只需FastMCP(name=..., auth=auth_provider)一行即可接入;客户端则通过auth="oauth"自动完成浏览器授权。无论是本地开发验证,还是接入企业 SSO 的生产部署,这条路径都可以直接复用。
进一步探索仓库可获得更完整的上下文:提供者实现见 fastmcp_slim/fastmcp/server/auth/providers/scalekit.py,JWT 验证核心见 fastmcp_slim/fastmcp/server/auth/providers/jwt.py,单元与集成测试见 tests/server/auth/providers/test_scalekit.py,官方集成文档见 docs/integrations/scalekit.mdx。
【免费下载链接】fastmcp🚀 The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考