MCP Python SDK 实战：OAuth 2.1 授权码全流程同宿主部署与 401 自动重试-迪斯科星球

MCP Python SDK 实战：OAuth 2.1 授权码全流程同宿主部署与 401 自动重试

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk

本指南基于当前仓库中的完整可运行示例 examples/stories/oauth/README.md（含配套的 client.py、server.py、server_lowlevel.py 及共享设施 examples/stories/_shared/auth.py），讲解如何用官方 MCP Python SDK 在单个 Starlette 应用内同时承载 OAuth 2.1 授权服务器（Authorization Server，AS）、受保护资源元数据（RFC 9728 PRM）与 Bearer 令牌门控的 MCP 端点，并让客户端OAuthClientProvider在第一次请求 401 后自动完成“PRM 发现 → AS 元数据发现 → 动态客户端注册（DCR）→ PKCE 授权码 → 令牌交换 → Bearer 重试”的完整链路。读完本文，你将掌握：一行构造器参数完成服务端 AS+RS 同宿主、客户端对Client(url)无auth=参数这一 SDK 空缺的补位方式，以及如何验证令牌与注册信息在重连后仍被复用。

一次请求内完成的 OAuth 流程全景

MCP 授权规范（对应 2025-11-25 版规范的 Authorization 章节）建立在 OAuth 2.1 之上：通过 Streamable HTTP 访问的 MCP 服务器是一个资源服务器（Resource Server，RS）——它验证令牌、拒绝无令牌请求，但本身不签发令牌；签发令牌是**授权服务器（AS）**的职责。

本示例的特殊之处在于把两者装进同一个进程、同一张 Starlette 应用：服务端一次MCPServer(auth=..., auth_server_provider=...)构造调用，就把下列路由全部挂到同一个 ASGI 应用上（路由实现在 src/mcp/server/auth/routes.py，其中定义了/authorize、/token、/register、/revoke等常量路径）：

RFC 9728 受保护资源元数据路由；
AS 路由：/register（DCR 动态注册）、/authorize、/token、/.well-known/oauth-authorization-server（AS 元数据发现）；
Bearer 令牌门控的/mcpMCP 端点。

客户端一侧，OAuthClientProvider是一个httpx2.Auth（见 src/mcp/client/auth/oauth2.py 中class OAuthClientProvider(RedirectAwareAuth)的实现，其requires_response_body = True，意味着它能读取首个 401 响应的响应体来执行发现流程）。它会对第一次请求收到的401做出反应，在首次被 await 的请求内部依次走完：

PRM 发现 → AS 元数据发现 → DCR 动态注册 → PKCE 授权码 → 令牌交换 → Bearer 重试

整个过程对上层业务代码完全透明：whoami的调用结果返回时，令牌已经就位，用户永远不会看到UnauthorizedError。

运行示例：三种启动方式

在仓库的 examples 目录下（stories包作为独立示例工程，见 examples/pyproject.toml，依赖mcp与 Python 3.10+），使用uv直接运行：

# HTTP —— 客户端自托管（self-host）同宿主的 AS + Bearer 门控 /mcp， # 在进程内无头完成授权码流程（重定向在进程内跟进），结束后拆除。 # 自托管使用本故事的固定端口 :8000（AS 元数据将其钉死），因此 :8000 必须空闲。 OAUTH_DEMO_AUTO_CONSENT=1 uv run python -m stories.oauth.client --http # 同上，但针对 lowlevel API 的服务端变体 OAUTH_DEMO_AUTO_CONSENT=1 uv run python -m stories.oauth.client --http --server server_lowlevel # 针对你自己启动的服务端（真实 uvicorn 跑在 :8000 上） OAUTH_DEMO_AUTO_CONSENT=1 uv run python -m stories.oauth.server --port 8000 & SERVER_PID=$! uv run python -m stories.oauth.client --http http://127.0.0.1:8000/mcp kill "$SERVER_PID"

端口为什么必须是 8000

示例的 AS 元数据（issuer、PRMresource）由_shared/auth.py中的同一个常量构建：

BASE_URL = "http://127.0.0.1:8000"
MCP_URL = f"{BASE_URL}/mcp"
REDIRECT_URI = f"{BASE_URL}/oauth/callback"

客户端与服务端都钉死在:8000。只要换端口，PRM/AS 发现链就会指向错误的 origin。这一点在 examples/stories/manifest.toml 的[story.oauth]条目中同样固化：fixed_port = 8000，注释写明“issuer/PRM metadata bake in :8000”。自托管时若 8000 端口已被占用，examples/stories/_harness.py 中的_self_hosted会直接抛出SystemExit提示你停掉占用进程或改用--http <url>直连，而不是挂死或误连到别人的服务上。

环境变量开关：`OAUTH_DEMO_AUTO_CONSENT`

设置OAUTH_DEMO_AUTO_CONSENT=1：演示 AS 跳过同意页（consent screen），直接 302 跳回并携带?code=...，从而让整个流程无头自动完成；
不设置：authorize步骤返回error=interaction_required，你可以借此观察在真实场景中浏览器本该打开的位置（examples/stories/_shared/auth.py 的InMemoryAuthorizationServerProvider.authorize在环境变量不等于"1"时调用construct_redirect_uri(target, error="interaction_required", state=params.state)）。

pytest 测试矩阵同样依赖该开关：manifest 中env = { OAUTH_DEMO_AUTO_CONSENT = "1" }会在每个测试分支执行前通过 monkeypatch 注入。

客户端实现解析

补位：`Client(url)`尚不支持`auth=`透传

当前 SDK 的Client(url)没有auth=透传参数，因此用裸 URL 构造的目标无法携带 OAuth 流程。示例通过两种 runner 以相同方式弥合这个缺口：模块导出build_auth，由_harness.run_client和 pytest 夹具据此构造一个已认证的httpx2.AsyncClient（将 provider 挂到其http.auth上），再通过_authed_targets让每个main收到的目标都已经过认证通道。相关逻辑见 examples/stories/_harness.py 的run_client与_authed_targets。

`build_auth()`：把 OAuth 挂到 httpx2 上

examples/stories/oauth/client.py 中的build_auth构造并返回一个OAuthClientProvider：

headless = HeadlessOAuth() headless.bind(http_client) return OAuthClientProvider( server_url=MCP_URL, client_metadata=OAuthClientMetadata( client_name="oauth-story-client", redirect_uris=[AnyUrl(REDIRECT_URI)], grant_types=["authorization_code", "refresh_token"], ), storage=InMemoryTokenStorage(), redirect_handler=headless.redirect_handler, callback_handler=headless.callback_handler, )

OAuthClientProvider的完整构造签名见 src/mcp/client/auth/oauth2.py 的__init__（server_url、client_metadata、storage、redirect_handler、callback_handler，以及可选的client_metadata_url与validate_resource_url）。要点如下：

server_url：MCP 端点地址，provider 从这里发现其余一切（PRM 元数据 → AS 元数据）；
client_metadata：OAuthClientMetadata是 RFC 7591 注册文档的 Pydantic 模型，构造时即校验（漏掉redirect_uris会在发出任何网络请求前抛出ValidationError），grant_types默认已是["authorization_code", "refresh_token"]，与本 provider 执行的流程完全一致；
storage：TokenStorage是一个只有四个异步方法的Protocol——get_tokens/set_tokens保存OAuthToken（访问令牌、刷新令牌、过期时间、scope），get_client_info/set_client_info保存 DCR 注册时 AS 签发的OAuthClientInformationFull（含client_id）。示例里的InMemoryTokenStorage仅把数据放在实例属性上，进程退出即遗忘；生产环境应持久化到文件或系统 keyring，且务必连client_info一起存——丢弃它会导致每次运行都重新动态注册；
redirect_handler/callback_handler：授权码流程中仅有的两处“人”的介入点。真实场景下前者负责把浏览器带到授权 URL，后者在用户回到redirect_uri后返回AuthorizationCodeResult。

第一次`Client`构造：整条流程在一个请求内完成

client.py的main中第一次Client(targets(), mode=mode)是整条流程发生的舞台：

async with Client(targets(), mode=mode) as client: first = await client.call_tool("whoami", {}) assert first.structured_content is not None assert "mcp" in first.structured_content["scopes"], first registered_id = first.structured_content["client_id"]

main收到的目标已带认证。首个请求打到线上后 401，OAuthClientProvider随即在结果到达业务代码之前走完 PRM 发现 → AS 元数据 → DCR → PKCE 授权码 → 令牌交换 → Bearer 重试。whoami返回的structured_content包含client_id与scopes，其中scopes断言包含"mcp"（与auth_settings(required_scopes=["mcp"])对应）。

第二次`Client`构造：令牌与注册复用的证明

Client在__aexit__之后不可再进入，重连意味着构造新的实例。第二次构造时，provider 的TokenStorage已持久化令牌与 DCR 注册信息，因此这个新连接从第一个请求起就携带Authorization: Bearer ...——没有第二次/authorize，也没有第二次/register：

async with Client(targets(), mode=mode) as reconnected: again = await reconnected.call_tool("whoami", {}) assert again.structured_content is not None assert again.structured_content["client_id"] == registered_id, again

由于演示 AS 每次 DCR 调用都会铸造全新的client_id，whoami两次返回相同的client_id就是注册与令牌确实被复用的直接证据。这也是TargetFactory（每次调用产生全新 target）配合multi_connection = true配置所验证的场景。

服务端实现解析

MCPServer：一次构造完成同宿主

examples/stories/oauth/server.py 的核心只有几行：

provider = InMemoryAuthorizationServerProvider() mcp = MCPServer( "oauth-example", auth=auth_settings(required_scopes=["mcp"]), auth_server_provider=provider, ) @mcp.tool(description="Return the authenticated principal's client_id and granted scopes.") def whoami() -> Principal: token = get_access_token() assert token is not None return Principal(client_id=token.client_id, scopes=token.scopes) return mcp.streamable_http_app(transport_security=NO_DNS_REBIND)

auth_server_provider=一参两职：provider 既是授权服务器（DCR/authorize/token 处理器），也是 Bearer 中间件校验令牌所依赖的令牌存储——InMemoryAuthorizationServerProvider在内部用三个字典（clients、codes、access_tokens）完成这一切；
不要同时传token_verifier=：MCPServer会从auth_server_provider派生令牌验证器，两者互斥，同时传入会触发互斥守卫（mutex guard）报错。官方文档 docs/run/authorization.md 也强调token_verifier=与auth=必须成对出现，否则构造时抛ValueError；
get_access_token()：whoami工具通过它读取已验证的主体信息。它由AuthContextMiddleware设置的每 HTTP 请求级contextvar 提供，而非 per-session 状态，实现位于 src/mcp/server/auth/middleware/auth_context.py；
auth_settings()：见 examples/stories/_shared/auth.py，它构建AuthSettings(issuer_url=AnyHttpUrl(BASE_URL), resource_server_url=AnyHttpUrl(MCP_URL), required_scopes=["mcp"], client_registration_options=..., validate_token_resource=True)——注意client_registration_options启用了 DCR 且限定合法 scope，validate_token_resource=True表示按 RFC 8707 校验资源指示器；
transport_security=NO_DNS_REBIND：由 examples/stories/_hosting.py 导出的TransportSecuritySettings(enable_dns_rebinding_protection=False)。仅用于本地演示，真实部署必须去掉（详见下文 Caveats）。

lowlevel 变体：同一条线，不同的接入点

examples/stories/oauth/server_lowlevel.py 展示同一应用的lowlevel.Server写法：auth=/token_verifier=/auth_server_provider=不再是构造器参数，而是streamable_http_app()的关键字参数：

server = Server("oauth-example", on_list_tools=list_tools, on_call_tool=call_tool) return server.streamable_http_app( auth=auth_settings(required_scopes=["mcp"]), token_verifier=ProviderTokenVerifier(provider), auth_server_provider=provider, transport_security=NO_DNS_REBIND, )

差别仅在接线位置：lowlevel 变体显式传入ProviderTokenVerifier(provider)（见 src/mcp/server/auth/provider.py），并用手工构造的types.Tool/types.ListToolsResult/types.CallToolResult返回结果。mcp.server.auth.*是一个 helper 层，lowlevel API 可以直接导入（本示例从 src/mcp/server/auth 导入了provider.ProviderTokenVerifier、middleware.auth_context.get_access_token、settings.AuthSettings等）。whoami工具的output_schema由模块级WHOAMI_OUTPUT_SCHEMA常量显式声明。

支撑设施：无头 OAuth 与自托管

`HeadlessOAuth`：进程内完成重定向

真实环境中，redirect_handler会打开浏览器，callback_handler会运行一个 loopback HTTP 监听器接收重定向。而演示用的 examples/stories/_shared/auth.py 中的HeadlessOAuth直接绑定到进程内的httpx2.AsyncClient：redirect_handler用follow_redirects=False请求授权 URL（注意必须传auth=None，否则会重入被锁定的认证流程造成死锁），断言返回 302 后从Location头解析出code与state；callback_handler直接返回该结果。这套“无头”方案只在演示 AS 自动同意（auto-consent）时才能成立。

`InMemoryAuthorizationServerProvider`：最小但完整的演示 AS

它继承OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]（基类契约见 src/mcp/server/auth/provider.py），实现了register_client、authorize、load_authorization_code、exchange_authorization_code、load_access_token等核心方法；load_refresh_token、exchange_refresh_token、revoke_token直接抛NotImplementedError（演示流程不涉及刷新与吊销）。authorize生成带 PKCEcode_challenge的授权码（5 分钟过期），exchange_authorization_code铸造 1 小时有效期的访问令牌并立即删除授权码（一次性使用）。

runner 与 manifest

examples/stories/_harness.py 的run_client负责解析命令行：--http后带 URL 则直连该 URL；裸--http则通过_self_hosted在子进程中拉起同目录的服务端并等待 TCP 就绪，退出时终止子进程。--server指定服务端变体（默认server，示例中即server_lowlevel的开关）。
examples/stories/manifest.toml 的[story.oauth]条目声明了该故事的运行矩阵：transports = ["http-asgi"]、server_export = "app"（导出 ASGI app）、multi_connection = true、fixed_port = 8000、env = { OAUTH_DEMO_AUTO_CONSENT = "1" }。测试入口 tests/examples/test_stories.py 依据这份 manifest 对每个故事执行 (transport × era × variant) 矩阵，并校验 manifest 与磁盘目录完全一致。

注意事项与边界

DNS 重绑定防护：transport_security=NO_DNS_REBIND关闭了 DNS-rebinding 防护——防护默认开启，而进程内 httpx2 桥接不发Origin头，所以演示必须关闭它；真实部署请移除该参数，保留默认防护。
无头 OAuth 的适用前提：HeadlessOAuth之所以能工作，完全依赖演示 AS 自动同意；真实场景下redirect_handler要打开浏览器，callback_handler要起 loopback HTTP 监听器接收重定向回跳。
导入路径较深：mcp.server.auth.*的导入路径目前较深（mcp.server还没有将其再导出为顶层命名空间），示例中均为直接深层导入。
演示 AS 的取舍：内存态令牌存储意味着重启即失效；刷新与吊销未实现。生产环境应使用真实 AS 与持久化存储。
示例仅限本地回环：auth_settings中issuer_url/resource_server_url使用http://127.0.0.1:8000，src/mcp/server/auth/routes.py 的validate_issuer_url也明确：RFC 8414 要求 HTTPS，仅对 loopback/localhost 放行明文 HTTP 用于测试。

企业官网建设流程全解析