mcp-agent OAuth 2.1 支持设计：从 Token 存储到委托授权流的完整实现指南

mcp-agent OAuth 2.1 支持设计：从 Token 存储到委托授权流的完整实现指南

【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent

导读

本文围绕 mcp-agent 仓库的 OAuth 支持设计文档（docs/oauth_support_design.md）展开，系统梳理该项目如何基于 OAuth 2.1、PKCE、RFC 8414/9728/8707/9068 以及 MCP 委托授权 SEP，为 "保护 MCP Agent Cloud 服务器" 与 "向受保护的下游 MCP 服务器认证" 两个方向提供完整闭环。读完本文，你将掌握 mcp-agent 的 OAuth 模块架构（src/mcp_agent/oauth/）、Token 存储与刷新机制、auth/request委托授权流程、回调处理与回环兜底方案，并能依据真实配置文件与示例（examples/oauth/）把 OAuth 集成到自己的 MCP 应用与工作流中。

一、设计目标：为什么要为 MCP Agent 引入 OAuth

设计文档开篇给出了四个核心目标，它们是理解整个模块的纲领：

1. 保护 MCP Agent Cloud 服务器：让 MCP 客户端通过标准 OAuth 2.1 流程获取 token，再以 Bearer Token 访问受保护的 MCP 服务；
2. 下游服务器认证：让 MCP Agent 运行时能够向需要 OAuth 访问令牌的下游 MCP 服务器完成认证；
3. 可插拔 Token 存储：本地开发用内存存储，多实例部署可切换 Redis（仓库当前已同时落地两者）；
4. 规范兼容性：对齐 MCP Authorization 规范（RFC 8414 授权服务器元数据、RFC 9728 受保护资源元数据、OAuth 2.1 + PKCE、Resource Indicators），并兼容提案中的 delegated authorization SEP。

从当前仓库源码看，这套设计已基本落地为独立包 src/mcp_agent/oauth/，并配套了四个测试文件（tests/test_token_manager.py、tests/test_token_verifier.py、tests/test_oauth_utils.py、tests/test_audience_validation.py）与三个可直接运行的示例（examples/oauth/interactive_tool、examples/oauth/pre_authorize、examples/oauth/protected_by_oauth）。

二、架构总览：六大组件与两类核心数据流

2.1 组件清单

设计文档规划的六大组件，与当前源码一一对应：

组件	职责	源码位置
Auth Server 集成	为 FastMCP 实例配置AuthSettings，并挂载自定义TokenVerifier调用鉴权服务	src/mcp_agent/server/app_server.py、src/mcp_agent/server/token_verifier.py
受保护资源元数据	通过 FastMCP hooks 提供/.well-known/oauth-protected-resource，供客户端发现授权服务器	src/mcp_agent/oauth/metadata.py（fetch_resource_metadata）
访问令牌校验	在每次入站 MCP 请求上强制 Bearer Token 校验，并把认证用户写入请求上下文	MCPAgentTokenVerifier+ FastMCPAuthSettings
OAuth Token 服务	mcp_agent.oauth包：TokenStore/TokenRecord、InMemoryTokenStore、Redis 实现、TokenManager、OAuthHttpxAuth、AuthorizationFlowCoordinator	src/mcp_agent/oauth/
委托授权 UI 流	网关/session relay 扩展，服务器可向 MCP 客户端发送auth/request，通过客户端回调 URL 或托管回调端点收取授权码	app_server.py中/internal/oauth/callback/{flow_id}路由与auth/request转发
配置面	扩展Settings与 per-server 的MCPServerAuthSettings，描述 scopes、首选授权服务器、redirect URI 及全局 token-store 配置	src/mcp_agent/config.py

设计文档在规划时把 Redis 实现标注为 "optional for multi-instance / follow-up"，但从源码看 src/mcp_agent/oauth/store/redis.py 中的RedisTokenStore已经完整落地，因此下文按实现现状讲解。

2.2 关键数据流：入站请求

1. 客户端携带 Bearer Token 发起请求；
2. MCPAgentTokenVerifier基于 RFC 9068（JWT Profile）语义进行校验，必要时先从/.well-known/oauth-authorization-server发现 introspection endpoint；
3. 校验通过后，token 被解析为携带身份信息的MCPAccessToken（见 src/mcp_agent/oauth/access_token.py），进而构造OAuthUserIdentity（provider + subject + email + claims，见 src/mcp_agent/oauth/identity.py）；
4. 该身份被传播到 workflows/sessions 上下文中，使下游 OAuth 流程能感知"正在操作的用户"。

OAuthUserIdentity的cache_key（provider:subject）是 token 按用户隔离存储的关键：TokenManager在获取 token 时按identity → resource → authorization_server → scopes逐级尝试候选身份（显式传入身份、当前上下文身份、session 身份、预配置默认身份），从而决定命中哪份缓存。

2.3 关键数据流：出站 HTTP（下游 MCP 服务器）

1. ServerRegistry检测到auth.oauth配置；
2. 将 HTTP transport 包装为OAuthHttpxAuth（见 src/mcp_agent/oauth/http/auth.py），它向TokenManager请求访问令牌；
3. TokenManager检查存储：缺失或过期时，由AuthorizationFlowCoordinator依次执行 RFC 9728 资源元数据发现、RFC 8414 授权服务器元数据发现、PKCE、委托浏览器流（经 MCP 客户端）或回环兜底、用授权码换 token、缓存结果；
4. 当响应返回 401/无效 token 时，OAuthHttpxAuth.async_auth_flow会先invalidate旧 token，再重新走ensure_access_token拿到刷新后的 token，复制原始请求体并自动重试一次。

OAuthHttpxAuth实现了httpx.Auth，其中requires_request_body = True，保证在重试时用request.copy()保留原始请求体，这是 401 自动重试正确性的关键细节。

2.4 关键数据流：Token 存储

• Token 以(user_identity, resource, authorization_server)为维度存储，并附带 metadata（scopes、expiry、refresh token、provider claims）；
• 存储键由 src/mcp_agent/oauth/store/base.py 中的TokenStoreKey定义，额外包含scope_fingerprint（对 scopes 去重排序后的确定性指纹），避免相同资源但不同 scope 的 token 相互污染；
• 存储实现提供乐观并发保护：TokenManager内部为每个TokenStoreKey维护独立的asyncio.Lock，防止并发刷新风暴；
• 可插拔后端：默认InMemoryTokenStore（store/in_memory.py），多实例部署用RedisTokenStore（store/redis.py）。

三、模块规划：mcp_agent.oauth包逐文件解读

设计文档给出了模块规划图，仓库中实际落地为：

src/mcp_agent/oauth/ __init__.py access_token.py # MCPAccessToken：带身份/claims 的扩展 token 模型 callbacks.py # OAuthCallbackRegistry：异步回调投递（flow_id / state 双通道） errors.py # 自定义异常层级 flow.py # AuthorizationFlowCoordinator identity.py # OAuthUserIdentity manager.py # TokenManager metadata.py # RFC 8414 / RFC 9728 元数据发现 pkce.py # PKCE + state 工具 records.py # TokenRecord store/base.py # TokenStore Protocol + TokenStoreKey store/in_memory.py # 默认内存存储 store/redis.py # Redis 多实例存储 http/auth.py # OAuthHttpxAuth

各模块要点如下：

• records.py：TokenRecord（pydantic BaseModel）承载 access_token、refresh_token、scopes、expires_at、token_type、resource、authorization_server 等字段；is_expired(leeway_seconds=...)支持刷新余量判断；with_tokens()用于刷新后原地更新令牌与获取时间。
• identity.py：OAuthUserIdentity为 frozen dataclass；from_access_token()从 introspection 结果构造身份；另有两个"合成身份"——DEFAULT_PRECONFIGURED_IDENTITY（无用户/session 时使用，标记token_source: synthetic）与session_identity()（按 session_id 生成确定性身份）。
• errors.py：异常层级为OAuthFlowError（基类）→AuthorizationDeclined、CallbackTimeoutError、TokenRefreshError、MissingUserIdentityError。
• pkce.py：generate_code_verifier（默认 64 字符，强制 43~128 的 RFC 7636 范围）、generate_code_challenge（SHA-256 + base64url，S256 方法）、generate_state（默认 32 字节 token_urlsafe）。
• metadata.py：fetch_resource_metadata（RFC 9728）、fetch_authorization_server_metadata（RFC 8414）、select_authorization_server（优先选择配置的首选服务器，否则取第一个）、normalize_resource（对 http/https URL 做 host 小写、去尾斜杠、去 query/fragment 的规范化）。

3.1 配置面：OAuth 相关 Settings 模型

设计文档要求扩展Settings与 per-server 认证配置，当前 src/mcp_agent/config.py 中已落地四个模型：

MCPOAuthClientSettings（下游服务器认证）关键字段：

字段	默认值	说明
enabled	False	是否对该下游服务器启用 OAuth
scopes	[]	授权时请求的 OAuth scopes
resource	None	RFC 8707 受保护资源标识（进入 authorize/token 请求）
authorization_server	None	授权服务器基础 URL，元数据由此 root 发现
client_id/client_secret	None	授权服务器注册的客户端凭据
access_token/refresh_token/expires_at/token_type	None	预配置 token，可绕过交互式流程
redirect_uri_options	[]	允许的 redirect URI 列表，流程从中选择
extra_authorize_params/extra_token_params	{}	追加到 authorize/token 请求的额外参数
require_pkce	True	是否强制 PKCE
use_internal_callback	True	是否优先使用内部回调 URL（否则走回环）
include_resource_parameter	True	是否在 authorize/token 请求中带resource参数

OAuthTokenStoreSettings（全局 token 持久化）：

字段	默认值	说明
backend	"memory"	memory或redis
redis_url	None	Redis 连接 URL
redis_prefix	"mcp_agent:oauth_tokens"	Redis key 前缀
refresh_leeway_seconds	60	到期前多少秒触发刷新

OAuthSettings（全局 OAuth 设置）：

字段	默认值	说明
token_store	OAuthTokenStoreSettings()	跨下游服务器共享的存储配置
flow_timeout_seconds	300（≥30）	授权回调等待上限
callback_base_url	None	内部回调基础 URL（use_internal_callback为 true 时使用）
loopback_ports	[33418, 33419, 33420]	纯客户端场景的回环端口候选列表，为空则禁用回环

MCPServerAuthSettings作为 per-server 的auth节点，包含api_key与oauth: MCPOAuthClientSettings | None。

3.2 服务器侧集成触点

设计文档列出的集成触点均已落地：

• src/mcp_agent/config.py：OAuth Settings 模型（见上）；
• src/mcp_agent/core/context.py：承载token_manager、token_store、oauth_config等上下文字段；
• src/mcp_agent/app.py：依据设置初始化 token store/manager；
• src/mcp_agent/server/app_server.py：create_mcp_server_for_app()在启用授权时构造AuthSettings(issuer_url=..., resource_server_url=..., required_scopes=...)并实例化MCPAgentTokenVerifier；同时注册内部回调路由/internal/oauth/callback/{flow_id}（GET/POST，include_in_schema=False），通过callback_registry.deliver(flow_id, payload)投递授权结果；relay 层在method == "auth/request"时把请求转发给上游 session（对应行 1084）；
• src/mcp_agent/server/token_verifier.py：MCPAgentTokenVerifier负责 introspection endpoint 的 well-known 发现（/.well-known/oauth-authorization-server）、token 校验与缓存、RFC 9068 audience 校验；
• src/mcp_agent/mcp/ 的mcp_server_registry.py与mcp_connection_manager.py：将OAuthHttpxAuth接入 HTTP transport，并提供手动 teardown 辅助。

四、OAuth 流程细节：从发现到刷新

设计文档将流程拆为五步，以下结合源码逐层展开。

4.1 发现（Discovery）

1. 若下游服务器响应 401 并带WWW-Authenticate，解析其中的resource_metadata，GET 资源元数据，进而确定授权服务器 URL；
2. 获取 RFC 8414 授权服务器元数据；
3. 在配置且受支持时执行可选的动态客户端注册。

TokenManager._resolve_oauth_context()（manager.py）实现了这一链路：

• resource 默认取oauth_config.resource，否则回退到服务器url，经normalize_resource规范化；
• 资源元数据发现会依次尝试/.well-known/oauth-protected-resource/{path}与/.well-known/oauth-protected-resource两个候选 URL；
• 授权服务器选择遵循select_authorization_server(resource_metadata, preferred)：优先匹配oauth_config.authorization_server，未命中则告警并回退第一个；
• 授权服务器元数据同样尝试/.well-known/oauth-authorization-server/{path}与/.well-known/oauth-authorization-server；
• 两类元数据都有 300 秒（5 分钟）的进程内缓存，避免每次请求都重复发现。

4.2 授权请求（Authorization Request）

AuthorizationFlowCoordinator.authorize()（flow.py）按顺序：

1. 校验user（无身份抛出MissingUserIdentityError）与client_id（缺失抛出OAuthFlowError）；
2. 组装 redirect 候选：优先use_internal_callback生成的内部回调{callback_base_url}/internal/oauth/callback/{flow_id}，其次追加回环候选（http://127.0.0.1:{port}/callback与http://localhost:{port}/callback，端口取loopback_ports，默认 33418/33419/33420），再合并redirect_uri_options；
3. 生成 PKCE verifier/challenge（S256）与安全随机state；
4. 构造授权 URL，参数含response_type=code、client_id、redirect_uri、scope、state、code_challenge、code_challenge_method=S256，并按include_resource_parameter追加 RFC 8707 的resource参数，最后并入extra_authorize_params；
5. 把完整的request_payload（含 url、message、redirect_uri_options、flow_id、scopes、timeout、token_endpoint、code_verifier、client_id、client_secret、issuer 等）通过_send_auth_request以auth/request发送给上游 MCP 客户端；
6. 若上游 session 不可用（抛出AuthorizationDeclined），则降级执行_run_loopback_flow（见下节）。

4.3 回调处理（Callback Handling）

• 首选路径：MCP 客户端通过请求结果返回回调 URL payload；协调器用_parse_callback_params解析 URL 的 query 与 fragment，提取code与error；
• 回退路径 A（内部托管回调）：授权服务器重定向到/internal/oauth/callback/{flow_id}，该路由（app_server.py）同时支持 GET/POST、query 参数与 JSON/form body，通过callback_registry把结果投递给等待中的 future；
• 回退路径 B（原生应用风格回环）：_run_loopback_flow在 127.0.0.1 上依序尝试绑定候选端口，成功后在本地启动asyncio.start_server临时 HTTP 监听器，用选定的 redirect_uri 重写授权 URL，打开浏览器（同时把完整 URL 打印到终端供手动访问），收到含code或error的回调后以state或flow_id路由投递。

回调结果统一进入校验：先检查error，再校验state与发起时的一致性（防 CSRF），最后提取code。

callback_registry（callbacks.py）是这套异步投递机制的核心：create_handle/deliver/deliver_by_state/fail/discard以asyncio.Future为单元，支持按 flow_id 与按 state 双通道投递，并在超时/异常时兜底清理。

4.4 Token 交换与存储（Token Exchange / Storage）

1. POST token endpoint，携带grant_type=authorization_code、code、redirect_uri、client_id、code_verifier，可选scope、resource（RFC 8707）与extra_token_params，confidential client 追加client_secret；
2. 解析响应：access_token、refresh_token、expires_in（换算为绝对时间戳）、scope（以响应为准回退到请求 scope）、token_type；
3. 组装TokenRecord并持久化；TokenManager.ensure_access_token在拿到新 token 后回填resource与authorization_server并写入 store。

另有两种"免交互"的 token 注入路径：

• 预配置 token（store_preconfigured_token）：当配置中直接给出access_token时，以DEFAULT_PRECONFIGURED_IDENTITY为身份写入 store，完全绕过交互流程；
• 工作流预授权（store_user_token）：通过workflows-store-credentials工具在异步工作流执行前缓存 token（详见 examples/oauth/pre_authorize/README.md），并校验调用方提供的authorization_server与解析出的 issuer 是否一致、scope 是否覆盖所需范围。

4.5 刷新与吊销（Refresh / Revocation）

• 刷新：TokenManager.get_access_token_if_present/ensure_access_token在record.is_expired(leeway_seconds=refresh_leeway_seconds)（默认 60 秒）时自动尝试_refresh_token：POSTgrant_type=refresh_token，携带 refresh_token、client_id、resource、scope（可选）与 client_secret；成功后以响应中的新 access_token/refresh_token/expires_in 生成新TokenRecord回写 store；
• 失效：刷新失败（TokenRefreshError）或 401 重试时调用invalidate()删除对应 key；对非默认身份，还会同步清理该身份下的默认身份副本；
• 吊销：设计文档要求"在授权服务器支持时提供吊销方法"，当前通过invalidate完成本地失效，服务端吊销接口取决于授权服务器能力（见开放问题）。

OAuthHttpxAuth将刷新与重试串成闭环：发送请求 → 401 → invalidate 旧 token → 重新ensure_access_token→request.copy()保留 body → 以新 token 重发。

五、并发安全：避免刷新风暴

设计文档与实现都强调了并发刷新控制。在TokenManager中：

• 每个TokenStoreKey对应一把asyncio.Lock（self._locks，defaultdict(asyncio.Lock)），任何 get/refresh/authorize 都在锁内进行；
• ensure_access_token在等待锁之后会"双重检查"（double-check）store，避免多个协程排队后重复发起授权流程；
• InMemoryTokenStore内部另有asyncio.Lock保护字典读写。

tests/test_token_manager.py中test_preconfigured_token_lookup_and_invalidation等用例直接验证了 store 的写入、命中与失效行为；设计文档规划的 "token store concurrency + expiry handling 单元测试" 对应tests/test_oauth_utils.py、tests/test_token_verifier.py、tests/test_audience_validation.py等测试套件。

六、实战示例与配置

examples/oauth/提供了三种互补场景（总览见 examples/oauth/README.md）。

6.1 interactive_tool：同步工具的完整授权码流

场景：MCP 服务器暴露github_org_search工具，首次调用时向客户端发送auth/request，客户端引导用户在浏览器完成 GitHub 登录；后续调用复用已缓存 token，无需再次弹窗。

前提（详见 examples/oauth/interactive_tool/README.md）：

1. 在 GitHub 创建 OAuth App，Authorization callback URL 必须精确填写http://127.0.0.1:33418/callback（与示例固定的回环端口一致）；
2. 由于 GitHub 不接受 RFC 8707resource参数，示例配置中需关闭include_resource_parameter；
3. 导出GITHUB_CLIENT_ID/GITHUB_CLIENT_SECRET，安装依赖pip install -e .。

运行：

# 终端一：启动服务器 python examples/oauth/interactive_tool/server.py # 终端二：运行客户端 python examples/oauth/interactive_tool/client.py

客户端展示授权提示 → 浏览器批准 → GitHub 重定向回本地回调处理器 → 工具结果打印在客户端终端。服务器与客户端使用稳定 session ID，首次授权后 token 被缓存并可跨运行复用。

6.2 pre_authorize：异步工作流的预授权

场景：Temporal 等后台执行的工作流无法自行交互式认证，因此在工作流运行前通过workflows-store-credentials工具播种 token（见 examples/oauth/pre_authorize/README.md）。

流程：

1. 复制 secrets 模板并填入 GitHub OAuth client_id/client_secret；
2. 导出已有 GitHub token：export GITHUB_ACCESS_TOKEN="github_pat_xxx"；
3. 启动服务器python examples/oauth/pre_authorize/main.py；
4. 另一终端运行python examples/oauth/pre_authorize/client.py：客户端先调用workflows-store-credentials缓存 token，再调用github_org_search工作流，后者通过gen_client("github", ...)访问 GitHub MCP 服务器。

关键配置片段（examples/oauth/pre_authorize/mcp_agent.config.yaml）：

oauth: loopback_ports: [33418, 33419, 33420] mcp: servers: github: transport: streamable_http url: "https://api.githubcopilot.com/mcp/" auth: oauth: enabled: true scopes: ["read:org", "public_repo", "user:email"] authorization_server: "https://github.com/login/oauth" use_internal_callback: false include_resource_parameter: false

main.py还展示了如何通过环境变量OAUTH_REDIS_URL切换到 Redis token store，并动态改写settings.oauth.token_store。

6.3 使用 Redis 持久化 token

默认 token 存内存，进程重启即丢失。改用 Redis 的完整步骤：

# 1. 启动 Redis docker run --rm -p 6379:6379 redis:7-alpine # 2. 安装可选依赖 pip install -e .[redis] # 3. 导出连接串 export OAUTH_REDIS_URL="redis://127.0.0.1:6379"

设置环境变量后，示例会自动切换到 Redis（key 前缀默认mcp_agent:oauth_tokens），服务器重启后 token 仍可复用。若代码中未设OAUTH_REDIS_URL，也可在配置中显式声明：

oauth: token_store: backend: redis redis_url: "redis://127.0.0.1:6379" redis_prefix: "mcp_agent:oauth_tokens" refresh_leeway_seconds: 60

RedisTokenStore用 URL-quote 后的user_key / resource / authorization_server / scope_fingerprint拼装 Redis key，以 JSON 序列化TokenRecord；redis包未安装时会提示pip install mcp-agent[redis]。

七、测试策略

设计文档规划的测试策略在仓库中已落地：

• token store 并发与过期处理：tests/test_token_manager.py（预配置 token 存取与失效、用户 token 的工作流/session 元数据、身份去重等）；
• 元数据发现 + PKCE 生成（纯 Python 测试）：tests/test_oauth_utils.py覆盖_candidate_resource_metadata_urls/_candidate_authorization_metadata_urls等候选 URL 构造逻辑，tests/test_token_manager.py亦直接导入这些函数；
• 服务器 401 强制 + WWW-Authenticate：tests/test_token_verifier.py与tests/test_audience_validation.py验证 token 校验、introspection 发现及 RFC 9068 audience 校验；
• 委托授权流的端到端验证（mocked HTTP + fake MCP client 确保auth/request管线贯通）由上述用例与示例配套覆盖。

八、开放问题与后续方向

设计文档明确列出的 follow-ups（当前仍待推进）：

1. 运维加固：token 轮换策略、速率限制；
2. MCPAgentTokenVerifier 定稿：LastMile 授权服务器如何暴露 token introspection + JWKS，需要具体 endpoint 规格才能定稿校验器实现；
3. auth/requestSEP 的客户端采纳度：需要能力检测（capability detection）；在客户端广泛支持之前，依赖托管回调回退与手动指引；
4. 访问控制 DSL：按 email/domain 的 include/exclude 规则，待 token 身份 payload 定稿后评估。

总结

mcp-agent 的 OAuth 支持是一套完整覆盖"保护自建 MCP 服务器"与"访问受保护下游 MCP 服务器"双向场景的实现：TokenRecord承载令牌，TokenStore抽象出可插拔的内存/Redis 持久化，TokenManager统一负责获取、刷新、失效与并发控制，AuthorizationFlowCoordinator打通了auth/request委托流、内部托管回调与 127.0.0.1 回环兜底，OAuthHttpxAuth则让下游 HTTP transport 获得"自动取 token、401 自动刷新重试"的能力。配合 examples/oauth/ 的交互式、预授权与 Redis 三组示例，开发者可以快速把 OAuth 2.1 + PKCE 的完整能力接入自己的 MCP 应用、异步工作流与多实例部署。

【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent