ADK Python MCP 工具集 OAuth 认证实战：adk-python 中 McpToolset 两阶段鉴权流程详解

ADK Python MCP 工具集 OAuth 认证实战：adk-python 中 McpToolset 两阶段鉴权流程详解

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

本文基于 adk-python 官方示例 mcp_toolset_auth，讲解 ADK（Agent Development Kit）的"工具集级认证"（toolset authentication）特性：当一个 MCP 服务器在列举工具和调用工具两个环节都要求 Bearer Token 时，McpToolset如何以"暂停—请求凭据—恢复"的两阶段流程完成 OAuth 认证。读完后，你将掌握两阶段认证事件的识别与处理、auth_scheme/auth_credential/AuthConfig三个核心对象的用法，以及如何在脚本或 ADK Web UI 中跑通一个受 OAuth 保护的 MCP 工具集。

一、工具集认证的两阶段模型

ADK 中工具的认证发生在两个时机：工具发现（向 MCP 服务器请求tools/list）和工具调用（tools/call）。普通的工具级认证（AuthenticatedFunctionTool等）只处理"调用时缺凭据"的情况，而这个示例演示的是更靠前的场景：在还没拿到工具列表之前，服务器就拒绝连接。

示例的 README 将该流程概括为两个阶段：

1. Phase 1（请求凭据）：Agent 在无凭据的情况下尝试从 MCP 服务器获取工具列表，工具集检测到需要认证，发出"需要认证"信号，返回一个认证请求事件（adk_request_credential函数调用）。
2. Phase 2（提供凭据后恢复）：用户提供 OAuth 凭据后，Agent 即可正常列举并调用工具。

从源码结构看，这套机制由两个部件支撑：

• 凭据 ID 前缀：auth_preprocessor.py 定义了TOOLSET_AUTH_CREDENTIAL_ID_PREFIX = "_adk_toolset_auth_"，注释明确写道："带此前缀的认证请求是工具集认证（发生在工具列举之前），且不需要恢复某个具体的函数调用"。这正是 README 中"credential ID 为_adk_toolset_auth_McpToolset，用以表明是工具集认证"的实现来源——前缀后拼接的是工具集类型名。
• 认证响应处理器：AuthPreprocessor在处理客户端回传的FunctionResponse时，会区分普通工具认证与工具集认证两类请求。后者不映射到某个可恢复的函数调用（auth_preprocessor.py 中第 3 步即"收集原始函数调用 ID 用于恢复，跳过工具集认证条目"），凭据存入会话状态后，下一次模型调用前工具集直接用新凭据重新列举工具。

与普通工具认证的关键区别在于：工具级认证的FunctionResponse必须携带等待中的工具调用的function_call.id，而工具集认证只认凭据 ID 的前缀标识，恢复后直接重跑工具发现，无需定位某个函数调用。

二、示例目录与文件职责

示例目录 contributing/samples/mcp/mcp_toolset_auth 包含三个核心文件（README 中的文件清单）：

文件	职责
oauth_mcp_server.py	要求 Bearer Token 认证的 MCP 服务器（FastMCP + FastAPI 中间件）
agent.py	配置了 OAuth 保护 MCP 工具集的 Agent
main.py	演示两阶段认证流程的测试脚本

运行方式（README 原文步骤，需在仓库根目录下执行）：

1. 在一个终端启动 MCP 服务器：

PYTHONPATH=src python contributing/samples/mcp/mcp_toolset_auth/oauth_mcp_server.py

2. 在另一个终端运行测试脚本：

PYTHONPATH=src python contributing/samples/mcp/mcp_toolset_auth/main.py

预期行为（README "Expected Behavior" 一节）：

1. 第一次调用产生一个adk_request_credential函数调用；
2. 凭据 ID 为_adk_toolset_auth_McpToolset，表明这是工具集认证；
3. 提供 access token 之后，Agent 即可列举并调用工具。

三、受保护 MCP 服务器的实现要点

oauth_mcp_server.py 构造了一个在列举工具和调用工具两个环节都校验 Authorization 头的服务器，这是触发"工具集认证"而非"工具调用认证"的关键——如果只在call_tool时校验，那么工具发现阶段不会失败，也就不会走到工具集认证分支。

3.1 认证校验逻辑

服务器定义了一个期望的测试令牌，并通过统一的校验函数处理：

# 期望的 OAuth 令牌（测试用） VALID_TOKEN = 'test_access_token_12345' def validate_auth_header(request: Request) -> bool: """Validate the Authorization header contains a valid Bearer token.""" auth_header = request.headers.get('authorization', '') if not auth_header.startswith('Bearer '): logger.warning('Missing or invalid Authorization header: %s', auth_header) return False token = auth_header[7:] # 去掉 'Bearer ' 前缀 if token != VALID_TOKEN: logger.warning('Invalid token: %s', token) return False return True

3.2 中间件：拦截所有 /mcp 请求

文件最巧妙的部分是用 FastAPI 中间件把认证检查置于 FastMCP 应用之前（oauth_mcp_server.py）。源码注释解释了原因：FastMCP 自己的 Starlette 应用才是真正服务/mcp端点的组件，把它挂到 FastAPI 应用下，才能让认证中间件拦截包括list_tools在内的每一个 MCP 请求：

mcp_app = mcp.streamable_http_app() # FastMCP 的会话管理器生命周期不随 mount 启动， # 所以在外层 FastAPI 的 lifespan 中手动 run @contextlib.asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: async with mcp.session_manager.run(): yield app = FastAPI(lifespan=lifespan) @app.middleware('http') async def auth_middleware(request: Request, call_next): """Middleware to validate auth on all MCP endpoints.""" if request.url.path.startswith('/mcp'): if not validate_auth_header(request): # 注意：这里选择返回 JSONResponse 而不是抛异常—— # HTTP 中间件里抛异常会绕过异常处理器，最终变成 500 return JSONResponse(status_code=401, content={'detail': 'Unauthorized'}) return await call_next(request) app.mount('/', mcp_app) if __name__ == '__main__': # 服务在 http://localhost:3001 uvicorn.run(app, host='localhost', port=3001)

两个值得注意的实现细节（均来自源码注释）：

• 返回 401 而非抛异常：从 HTTP 中间件中抛出的异常会绕过 FastAPI 的异常处理器，被 uvicorn 直接翻译成 500 错误。对客户端而言，401 才是"需要认证"的语义信号。
• 兼容 MCP SDK 两个大版本：MCP 2.0 将FastMCP所在的模块从mcp.server.fastmcp迁移到mcp.server.mcpserver，且run()的绑定参数从构造器移到了别处。示例用try/except ImportError同时兼容两个版本（oauth_mcp_server.py），并以uvicorn.run()显式绑定地址而非mcp.run()。

服务器暴露两个工具：get_user_profile(user_id)与list_users()，返回 Alice / Bob 两个模拟用户。

四、Agent 配置：auth_scheme、auth_credential 与 McpToolset

agent.py 展示了认证三元组如何装配到McpToolset。

4.1 声明 OAuth2 授权码方案

auth_scheme使用fastapi.openapi.models中的 OAuth2 模型，声明授权端点、令牌端点和所需 scope：

# OAuth2 授权码流程，声明完整 OAuth 流程所需的元数据 auth_scheme = OAuth2( flows=OAuthFlows( authorizationCode=OAuthFlowAuthorizationCode( authorizationUrl='https://example.com/oauth/authorize', tokenUrl='https://example.com/oauth/token', scopes={'read': 'Read access', 'write': 'Write access'}, ) ) )

4.2 声明原始凭据

auth_credential是"原始凭据"（raw credential），即用于换取令牌的 OAuth 客户端身份。注意示例中它只含client_id/client_secret，尚不含 access token：

auth_credential = AuthCredential( auth_type=AuthCredentialTypes.OAUTH2, oauth2=OAuth2Auth( client_id='test_client_id', client_secret='test_client_secret', ), )

这两个类的语义在 auth_tool.py 的AuthConfig字段文档中有权威定义：raw_auth_credential是"用于收集凭据的原始凭据，在需要凭据交换的方案（如 OAuth2、OIDC）中使用"；exchanged_auth_credential是"由 ADK 与客户端协作填写的工作副本"。本示例走的是"客户端直接给出 access token"的路径，exchanged_auth_credential由 main.py 在恢复阶段填充。

4.3 组装 McpToolset 与 LlmAgent

mcp_toolset = McpToolset( connection_params=StreamableHTTPConnectionParams( url='http://localhost:3001/mcp', ), auth_scheme=auth_scheme, auth_credential=auth_credential, ) root_agent = LlmAgent( name='oauth_mcp_agent', instruction="""You are a helpful assistant that can access user information. ...""", tools=[mcp_toolset], )

对照 mcp_toolset.py 的构造函数可以确认两件事：

• auth_scheme与auth_credential是McpToolset的一级参数，分别对应"工具调用时的认证方案"与"认证凭据"。
• 当提供了auth_scheme时，构造函数内部会自动构造一个AuthConfig并保存在self._auth_config上，源码注释说明其用途："ADK 会在调用get_tools()之前，往这个 config 中填充exchanged_auth_credential，工具集随后即可通过self._auth_config.exchanged_auth_credential访问到可直接使用的凭据"（mcp_toolset.py）。

这正是两阶段流程的衔接点：凭据到达后，框架就地更新工具集内部的AuthConfig，下一次get_tools()就能带上 Authorization 头。

4.4 请求头如何拼装

McpToolset._build_headers（mcp_toolset.py）负责为每个 MCP 会话合并两类头：header_provider回调产出的自定义头，以及由凭据与方案计算出的认证头。后者的入口_get_auth_headers优先从ReadonlyContext中按credential_key取凭据，取不到再回退到self._auth_config.exchanged_auth_credential，最终交给build_auth_headers(credential, auth_scheme)生成。对于 OAuth2 方案，结果就是Authorization: Bearer <access_token>头——与服务器端validate_auth_header的期望一一对应。

五、main.py：两阶段认证驱动脚本

main.py 是整个示例的操作核心，完整演示了"首次调用触发认证请求 → 回传凭据 → 恢复调用"的闭环。

5.1 Phase 1：无凭据发起请求

脚本创建InMemorySessionService和Runner后，以 "List all users" 作为用户消息发起第一次调用，并逐事件检查是否出现adk_request_credential函数调用，同时记录其调用 ID（后续恢复时必须匹配）：

async for event in runner.run_async( session_id=session.id, user_id='test_user', new_message=types.Content( role='user', parts=[types.Part(text=user_message)], ), ): if part.function_call and \ part.function_call.name == 'adk_request_credential': auth_function_call_id = part.function_call.id

若脚本捕获到异常，会提示"Make sure the MCP server is running!"——因为 Phase 1 失败最常见的原因就是服务器未启动（连接错误与 401 都可能导致列举失败）。

5.2 Phase 2：构造 AuthConfig 并回传

检测到认证请求后，脚本模拟用户完成 OAuth 流程，构造AuthConfig响应。其中exchanged_auth_credential直接填入了服务器校验通过的 access token：

auth_response = AuthConfig( auth_scheme=auth_scheme, raw_auth_credential=auth_credential, exchanged_auth_credential=AuthCredential( auth_type=AuthCredentialTypes.OAUTH2, oauth2=OAuth2Auth( access_token='test_access_token_12345', ), ), )

随后把它序列化为FunctionResponse，作为一条新的 user 消息发起第二次调用：

auth_response_message = types.Content( role='user', parts=[ types.Part( function_response=types.FunctionResponse( name='adk_request_credential', id=auth_function_call_id, response=auth_response.model_dump(exclude_none=True), ) ) ], ) async for event in runner.run_async( session_id=session.id, user_id='test_user', new_message=auth_response_message, ): ...

这里有两个容易踩坑的细节：

• id必须复用 Phase 1 中adk_request_credential函数调用的 ID。处理器正是靠这个 ID 把客户端响应与服务器当初发出的认证请求配对（参见 _store_auth_and_collect_resume_targets：它先扫描会话事件找出匹配的adk_request_credential调用，再把响应里的凭据回填到原始请求的AuthConfig上）。
• model_dump(exclude_none=True)：AuthConfig通过 pydantic alias 以 camelCase 键序列化（如exchangedAuthCredential），这也是 tool_auth 文档中"keys 为 camelCase 因为 config 是按 alias dump 的"的由来。

脚本结尾显式调用await mcp_toolset.close()释放 MCP 会话，与McpToolset.close()文档中"关闭 MCP 会话并清理相关资源，可安全地多次调用"的约定一致（mcp_toolset.py）。

5.3 恢复阶段的安全校验

从 auth_preprocessor.py 的实现可以看到恢复阶段对客户端回传内容的防御姿态：认证方案、原始凭据与凭据键一律以服务器当初发出的请求为准，客户端响应只贡献"浏览器往返的结果"（即交换后凭据），且通过_merge_credential_oauth2_fields逐字段回填——注释直言原因："方案指定了开发者密钥将被投递到的令牌端点，原始凭据指定了以哪个 OAuth2 客户端身份投递，所以两者必须保持本服务器发出的原样"。对无法匹配到会话内请求的响应 ID，处理器会记录告警并忽略，避免恶意客户端借认证响应选择任意的令牌端点。

六、用 ADK Web UI 测试

除脚本方式外，README 还给出了 Web UI 的测试方式：

adk web contributing/samples/mcp/mcp_toolset_auth

README 的提示原文：Web UI 会展示认证请求（auth request），你需要手动提供凭据。由于本示例的 MCP 服务器并不提供真实的 OAuth 授权端点（example.com仅为占位），在 Web UI 中你走的是"直接提供 access token"的路径：UI 捕获到adk_request_credential事件后，将 token（test_access_token_12345）填入响应即可完成 Phase 2，效果与 main.py 的自动化流程等价。

七、要点回顾与延伸阅读

本示例的价值在于它把 ADK 认证体系中一个容易被忽略的场景讲清楚了：认证可以发生在工具发现之前。核心结论：

1. 两阶段的触发条件是"工具列举也需要凭据"。McpToolset构造AuthConfig存入_auth_config，工具列举失败于 401 时，LLM flow 发出adk_request_credential事件并将调用暂停（base_llm_flow.py 同样定义了该前缀常量，保证 flow 与 preprocessor 判断一致）。
2. 凭据 ID 前缀_adk_toolset_auth_是工具集认证的识别标志，恢复时无需匹配具体的等待函数调用。
3. 客户端回传凭据时，FunctionResponse.id必须等于adk_request_credential调用的 ID，且响应消息需作为最新一条 user 事件发送。
4. 服务器侧用中间件返回 401（而非抛异常）+ 兼容 MCP SDK 1.x/2.x 的导入方式，是让示例跨环境可运行的关键工程细节。

如需进一步了解 ADK 认证机制的整体模型（AuthScheme/AuthCredential的类型系统、CredentialManager的凭据查找顺序、凭据服务如何避免重复授权），可参考 tool_auth 官方指南——该指南的"Related samples"一节将本示例列为工具集级认证的代表性参考。其他 MCP 相关示例（SSE、Stdio、动态请求头、服务端采样等）位于 contributing/samples/mcp。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python