1. 从零理解 ADK 与 MCP 协同的智能代理工作流如果你最近在折腾 Python 智能代理大概率会同时刷到 ADK 和 MCP 这两个词。ADK 全称 Agent Development Kit是 Google 开源的一套以代码为中心的 Python 工具包专门用来构建、评估和部署智能代理MCP 全称 Model Context Protocol是一套开放协议用来标准化大模型与外部工具、数据源之间的交互方式。把这两个东西拼在一起你就能得到一个既能推理规划、又能真正调用外部工具的 Agent 工作流。这篇文章面向的是想从零搭建可运行 Agent 的 Python 开发者。我会带你走完一条完整链路初始化 ADK 项目、接入一个符合 MCP 标准的服务、通过统一的 Key/API 通道完成模型调用配置最后用一次端到端任务执行来验证代理是否真的可用。整个过程不需要你预先理解 ADK 的全部概念跟着步骤走就能跑起来。先说清楚 ADK 里的几个核心角色不然后面看代码会懵。ADK 提供三类代理LLM 代理比如 LlmAgent负责用大模型做理解、推理、规划和行动工作流代理SequentialAgent、ParallelAgent、LoopAgent用可预测的模式编排其他代理不依赖 LLM 做流程控制自定义代理则通过扩展 BaseAgent 实现特殊逻辑。我们这篇用的是 LLM 代理配合 MCP 工具这是最容易落地、也最贴近真实业务的一种组合。那工具在 ADK 里是什么工具就是授予代理的特定能力让它能执行动作、和外部世界交互而不只是生成文本。工具可以是普通 Python 函数、类方法甚至另一个代理。代理通过函数调用机制动态使用工具LLM 对上下文推理选择合适的工具生成输入参数观察返回结果再把结果整合进下一步动作或最终回复。ADK 支持函数工具、内置工具网络搜索、代码执行、RAG 等、第三方工具LangChain、CrewAI 生态以及我们重点要用的 MCP 工具。MCP 工具和 ADK 原生工具的区别在于来源。ADK 工具是直接写在代理项目里的 Python 对象MCP 工具由独立的 MCP 服务器暴露ADK 通过 MCPToolset 把它适配成代理可调用的工具。这种解耦的好处是工具可以独立开发、独立部署、跨项目复用代理只负责发现和调用。对于需要接入实时数据比如航班、天气、库存的场景MCP 服务器是更干净的边界。架构上一次完整的调用是这样流动的用户输入进入 RunnerRunner 协调会话和代理执行LlmAgent 把可用工具列表连同用户意图一起交给大模型模型决定调用哪个 MCP 工具MCPToolset 通过 StdioServerParameters 启动或连接 MCP 服务器进程工具执行后把结果回传给模型模型再生成最终回复。整条链路是异步的因为 ADK 和 MCP 的 Python 库都构建在 asyncio 之上。理解了这条链路你就知道为什么配置环节一个都不能少模型通道决定 LLM 能不能被调用MCP 服务器决定工具能不能被发现会话服务决定上下文能不能被追踪。下面我按顺序把每一步都拆开讲代码可以直接复制。2. TaoToken 前置统一 Key 与 API 通道配置在真正写代理代码之前得先把模型调用通道打通。ADK 默认期望你配置 Google 的访问凭证但实际开发中我们往往希望用一个统一的入口来管理 Key 和 API 地址避免在多个项目里散落不同的凭证配置。TaoToken 在这里扮演的就是统一通道的角色你拿到一个 Key配好 Base URL就能在 ADK 项目里完成模型调用不用为每个环境单独折腾。先明确三个必须对齐的要素Base URL、API Key、Model ID。这三件套在任何接入场景里都是核心缺一个都会报错。TaoToken 的 API 地址是 https://taotoken.net/api注意这个地址不带任何查询参数直接作为基础端点使用。Key 需要你在控制台里创建创建入口在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 进去之后新建一个 API Key复制出来保存好后面配置环境变量要用。模型 ID 这块要看你实际想调用的模型。ADK 的 LlmAgent 接受一个 model 参数你可以把它设成你账号下可用的模型标识。建议先在模型对话页面确认一下你的 Key 能正常调用哪些模型入口是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 在那边发一条测试消息确认通道通了再回到代码里配置能省掉很多排查时间。环境变量怎么设ADK 里模型凭证的变量名是 GOOGLE_API_KEY这一点和直接用 Gemini SDK 时的 GEMINI_API_KEY 不一样很多人第一次踩坑就踩在这里。你需要在 shell 里导出export GOOGLE_API_KEY你的TaoToken Key export GOOGLE_GENAI_USE_VERTEXAIfalse export SERP_API_KEY你的SerpAPI Key第二行很关键。ADK 默认会尝试走 Vertex AI 路径如果你没有配置 GCP 项目、位置这些参数它会直接报缺少 key inputs 的错误。把 GOOGLE_GENAI_USE_VERTEXAI 显式设为 false强制走 API Key 路径就能避开这个坑。第三行的 SERP_API_KEY 是给后面 MCP 航班搜索工具用的如果你暂时不接那个工具可以先不设。如果你更习惯用配置文件而不是环境变量也可以在项目根目录建一个 .env 文件把上面三行写进去然后在代码入口用 python-dotenv 加载。不过要注意ADK 的某些组件在子进程里启动 MCP 服务器时环境变量需要显式传递所以纯 .env 方案在涉及 StdioServerParameters 的场景下不一定够用稳妥做法还是 shell 导出加代码里显式传 env。配置完成后建议先做一次最小验证写一个几行的 Python 脚本用你配置的通道发一条最简单的请求确认返回正常。这一步不做后面代理跑不起来你会分不清是通道问题还是代码问题。验证通过后再进入 ADK 项目初始化节奏会顺很多。关于 Key 的安全管理别把 Key 硬编码进代码提交到仓库。用环境变量或者本地未跟踪的配置文件团队协作时通过各自的 shell 配置或密钥管理工具注入。TaoToken 控制台里可以随时吊销和重建 Key所以万一泄露了也有补救手段但养成不硬编码的习惯能省掉很多麻烦。3. 可复制配置ADK 项目初始化与 MCP 接入现在进入动手环节。先建虚拟环境、装依赖这一步在 mac 或 Unix 上python -m venv venv source venv/bin/activate pip install google-adk pip install mcp-flight-search pip install google-genai三个包各司其职google-adk 是代理开发工具包mcp-flight-search 是一个用 FastMCP 构建的轻量 MCP 服务器通过 SerpAPI 暴露航班搜索工具google-genai 是和大模型交互的 SDK。装完之后你的项目目录结构建议这样组织adk-mcp-agent/ ├── agent.py ├── .env └── requirements.txt接下来是核心代码。先看 MCP 工具获取部分这里用 StdioServerParameters 定义如何启动 MCP 服务器进程import os import asyncio from google.adk.agents.llm_agent import LlmAgent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters from google.genai import types async def get_tools_async(): 从航班搜索 MCP 服务器获取工具。 print(尝试连接到 MCP 航班搜索服务器...) server_params StdioServerParameters( commandmcp-flight-search, args[--connection_type, stdio], env{SERP_API_KEY: os.getenv(SERP_API_KEY)}, ) tools, exit_stack await MCPToolset.from_server( connection_paramsserver_params ) print(MCP 工具集创建成功。) return tools, exit_stack这段代码里command 指定要启动的可执行程序args 是传给它的参数env 是子进程的环境变量。MCPToolset.from_server 会异步连接服务器、拉取工具列表并返回一个 exit_stack 用于后续清理连接。exit_stack 千万别丢否则 MCP 服务器进程可能残留。然后是创建代理async def get_agent_async(): 创建一个配备 MCP 工具的 ADK 代理。 tools, exit_stack await get_tools_async() print(f从 MCP 服务器获取了 {len(tools)} 个工具。) root_agent LlmAgent( modelos.getenv(GEMINI_MODEL, gemini-2.0-flash), nameflight_search_assistant, instruction帮助用户使用可用工具根据提示搜索航班。如果未指定返回日期则使用空字符串表示单程旅行。, toolstools, ) return root_agent, exit_stackmodel 这里我用了 gemini-2.0-flash原因是免费层级的速率限制比较友好pro 系列在测试阶段很容易撞 429。name 是代理标识instruction 是系统提示tools 直接接收从 MCP 拿到的工具列表。ADK 会自动把这些 MCP 工具适配成代理可调用的格式。最后是编排和运行async def async_main(): session_service InMemorySessionService() session session_service.create_session( state{}, app_nameflight_search_app, user_iduser_flights ) query Find flights from Atlanta to Las Vegas 2025-05-05 print(f用户查询{query}) content types.Content(roleuser, parts[types.Part(textquery)]) root_agent, exit_stack await get_agent_async() runner Runner( app_nameflight_search_app, agentroot_agent, session_servicesession_service, ) print(运行代理...) events_async runner.run_async( session_idsession.id, user_idsession.user_id, new_messagecontent ) async for event in events_async: print(f收到事件{event}) print(关闭 MCP 服务器连接...) await exit_stack.aclose() print(清理完成。) if __name__ __main__: asyncio.run(async_main())InMemorySessionService 把所有会话数据放在内存里应用重启就丢适合本地验证。Runner 负责协调代理生命周期run_async 返回一个异步事件流你可以逐个消费事件观察代理的思考过程。最后一定要调 exit_stack.aclose()否则 MCP 子进程不会退出。如果你用的是 Cline MCP 或 Claude Code 这类客户端来管理 MCP 服务器配置格式会不一样但三件套不变Base URL、Key、Model ID。以 Cline 的 MCP 配置为例通常是一个 JSON{ mcpServers: { flight-search: { command: mcp-flight-search, args: [--connection_type, stdio], env: { SERP_API_KEY: 你的SerpAPI Key } } } }注意这里的 env 只放 MCP 服务器自己需要的凭证模型通道的 Key 是在 ADK 侧配置的两者不要混。如果你在 Codex 里用 auth.json 管理凭证也是同样的原则模型通道和工具通道分开配置各管各的。4. 验证请求端到端任务执行与结果确认配置写完了现在跑一次端到端任务确认代理真的能用。执行python agent.py正常情况下你会看到类似这样的输出尝试连接到 MCP 航班搜索服务器... MCP 工具集创建成功。 从 MCP 服务器获取了 1 个工具。 用户查询Find flights from Atlanta to Las Vegas 2025-05-05 运行代理... 收到事件... 关闭 MCP 服务器连接... 清理完成。事件流里会包含模型的推理过程、工具调用请求、工具返回结果、以及最终生成的回复。如果你看到工具被调用并且返回了航班数据说明整条链路通了模型通道正常、MCP 服务器正常、工具适配正常、会话管理正常。想看得更清楚可以把事件打印改成结构化输出只提取关键字段async for event in events_async: if event.content and event.content.parts: for part in event.content.parts: if part.text: print(f[文本] {part.text}) if part.function_call: print(f[调用工具] {part.function_call.name} 参数{part.function_call.args}) if part.function_response: print(f[工具返回] {part.function_response.name})这样你能清晰看到代理先决定调用哪个工具、传了什么参数、拿到什么结果、最后怎么组织回复。调试阶段这个输出比原始事件流有用得多。验证成功的标志有三个一是没有抛异常二是日志里出现了工具调用记录三是最终回复里包含了基于工具返回数据的内容。三个都满足说明你的 ADK MCP 工作流已经可运行了。如果只满足前两个但最终回复是空的通常是模型通道的问题回去检查 GOOGLE_API_KEY 和 GOOGLE_GENAI_USE_VERTEXAI 这两个变量。再补一个验证技巧把 query 换成一个不需要工具就能回答的问题比如「你好介绍一下你自己」如果这个能正常回复说明模型通道没问题问题就缩小到 MCP 工具侧了。这种二分排查法在代理调试里特别省时间。5. 本篇常见错排查401、429、500 与连接失败跑不通的时候报错信息往往不会直接告诉你根因。下面是我实际遇到过的几类典型错误和对应解法。第一类ValueError: Missing key inputs argument。完整报错类似ValueError: Missing key inputs argument! To use the Google AI API, provide (api_key) arguments. To use the Google Cloud API, provide (vertexai, project location) arguments.这个错误的根因是 ADK 默认走 Vertex AI 路径而你没配 GCP 项目。解法是显式设置 GOOGLE_GENAI_USE_VERTEXAIfalse并确保 GOOGLE_API_KEY 已经导出。注意变量名是 GOOGLE_API_KEY不是 GEMINI_API_KEY写错了错误信息不会明确提示你只会说缺 key。第二类401 未授权。这个通常出现在 Key 无效、过期或者复制时带了空格。检查方法是在模型对话页面用同一个 Key 发一条消息如果那边也 401就是 Key 本身的问题去控制台重新创建一个。如果那边正常但代码里 401检查环境变量有没有被正确加载特别是子进程场景下 env 有没有传对。第三类429 RESOURCE_EXHAUSTED。完整报错类似google.genai.errors.ClientError: 429 RESOURCE_EXHAUSTED. {error: {code: 429, message: You exceeded your current quota...}}这是速率限制。免费层级配额低pro 系列模型尤其容易撞。解法是切换到 flash 系列模型把 model 参数改成 gemini-2.0-flash。如果业务确实需要 pro就得升级到付费层级或者做请求节流。第四类500 INTERNAL。报错类似An error occurred during execution: 500 INTERNAL. {error: {code: 500, message: An internal error has occurred...}}这类错误多数是服务端临时问题重试往往能过。如果持续出现检查你的请求里有没有超长上下文或者异常参数。代理场景下工具返回的数据如果特别大也可能触发这类错误可以在工具侧做结果截断。第五类local proxy failed 或连接超时。这类错误通常和网络环境有关检查你的出口网络是否稳定以及 Base URL 是否配置正确。TaoToken 的 API 地址是 https://taotoken.net/api不要多加路径或者参数。如果用了自定义的 HTTP 客户端确认没有额外的代理配置干扰。第六类MCP 服务器启动失败。表现是 get_tools_async 卡住或者抛异常。检查 mcp-flight-search 是否装在了当前虚拟环境里command 路径是否可执行env 里的 SERP_API_KEY 是否有效。可以在 shell 里直接跑一次 mcp-flight-search --connection_type stdio 看能不能启动排除环境问题。第七类OAuth 相关报错。如果你在别的工具里配了 OAuth 流程报错里出现 OAuth 字样通常是凭证刷新失败或者 scope 不对。这类问题在纯 API Key 场景下不会出现如果你没用到 OAuth 却看到这个报错检查是不是某个客户端配置残留了旧的认证方式。排查的通用思路是分层定位先确认模型通道用最小请求验证再确认 MCP 服务器单独启动验证最后确认代理编排换简单 query 验证。每一层单独通了组合起来才不会互相甩锅。6. 语义一致 CTA把工作流接到你的真实项目跑通这个最小示例之后下一步就是把它接到你自己的业务里。几个实用的延伸方向把 mcp-flight-search 换成你自己的 MCP 服务器暴露你业务需要的工具把 InMemorySessionService 换成持久化的会话服务让上下文跨重启保留把单代理扩展成多代理编排用 SequentialAgent 或 ParallelAgent 处理更复杂的流程。模型通道这块如果你要长期跑编码类或 Agent 类任务可以了解一下 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 适合需要稳定调用额度的场景。如果只是偶尔验证模型效果用模型对话页面就够了。接入过程中遇到配置问题接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有各语言和各客户端的配置示例。Key 的管理和创建统一在 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。最后分享一个我踩过的坑MCP 服务器是有状态的长连接和传统无状态 REST API 不一样。这意味着在远程部署、多用户场景下会话亲和性和连接生命周期管理会变成新的复杂度来源。本地验证阶段用 StdioServerParameters 没问题但上生产前一定要想清楚连接怎么复用、怎么清理、怎么在多个用户之间隔离。这个坑不提前想后面扩展时会很痛。