1. 企业级 Agent 为什么需要 MCP 中间层
如果你正在做企业级 Agent 架构,大概率会遇到这样一个尴尬局面:Agent 数量从 3 个涨到 30 个,工具从 5 个涨到 50 个,代码库开始失控。每接一个新工具,就要在 Agent 里写一遍适配逻辑;每换一个模型供应商,Function Calling 的 schema 又要重写一遍。这就是典型的 N×M 扩展性问题——Agent 数量乘以工具数量,复杂度指数级上升。
MCP(Model Context Protocol)中间层的价值就在这里。它把 Agent 和工具彻底解耦:Agent 只认 MCP 协议,工具只暴露 MCP Server 接口,双方通过标准化的 JSON-RPC 通信。这样一来,新增工具不用改 Agent 代码,替换 Agent 框架也不用重写工具层。企业级场景下,MCP 中间层还承担了权限管控、调用审计、限流熔断这些职责,相当于给 Agent 和工具之间加了一道可控的网关。
这篇文章我会用 Google ADK 作为 Agent 框架、FastMCP 作为 Server 实现,搭一个可运行的文件系统访问中间层原型。同时把 LLM 接入通道统一到 TaoToken,这样你只需要维护一套 Key 和 API 地址,就能在 ADK、Coding Plan、模型对话之间自由切换。整套代码可以直接收藏,改改路径就能跑。
2. TaoToken 前置准备:统一 Key 与 API 通道
在动手写 MCP Server 之前,先把 LLM 接入通道固定下来。企业级项目最怕的就是每个模块各接一套模型 API,Key 散落在各个配置文件里。我的做法是统一走 TaoToken 的 API 通道,ADK 里的 LiteLLM 只需要指向一个 base_url。
第一步,登录 TaoToken 控制台创建 API Key。地址是 https://taotoken.net/api-keys ,创建后复制保存,后面 config.toml 和 settings.json 都要用。
第二步,确认 API 基础地址。TaoToken 的 API 入口是 https://taotoken.net/api ,兼容 OpenAI 格式,所以 LiteLLM 可以直接用 openai/ 前缀调用。如果你用的是 Claude 系列模型,走 https://taotoken.net/api 的 Anthropic 兼容端点即可。
第三步,把 Key 写进环境变量,不要硬编码在代码里。Linux/macOS 下:
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"Windows PowerShell:
$env:TAOTOKEN_API_KEY="sk-你的key" $env:TAOTOKEN_BASE_URL="https://taotoken.net/api"这样 ADK 里的 LiteLLM 配置就能直接读取环境变量,MCP Server 本身不需要感知 Key,职责边界清晰。如果你还没决定用哪个模型,可以先去 https://taotoken.net/models 看看模型对话能力,确认工具调用(Function Calling)支持情况再定。
3. 可复制配置:config.toml 与 settings.json
企业级项目讲究配置与代码分离。我把 MCP Server 的启动参数、ADK 的模型配置、TaoToken 的接入信息分别放在 config.toml 和 settings.json 里,这样换环境不用改代码。
先看 config.toml,放在项目根目录:
# config.toml - MCP 中间层配置 [mcp_server] name = "filesystem-mcp" transport = "sse" host = "0.0.0.0" port = 3000 sse_path = "/sse" [mcp_server.security] # 企业级必配:允许访问的根目录白名单 allowed_roots = ["/data/workspace", "/tmp/agent_sandbox"] # 单次调用超时(秒) call_timeout = 30 # 是否开启调用审计日志 audit_log = true [llm] provider = "openai" model = "deepseek-chat" base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" temperature = 0.2 max_tokens = 2048 [agent] name = "file_agent" description = "自然语言驱动的文件访问 Agent" max_tool_calls = 5再看 settings.json,这是 ADK 读取的 Agent 配置:
{ "agent": { "name": "file_agent", "model": "openai/deepseek-chat", "instruction": "你是一个文件管理助手,通过 MCP 工具访问文件系统。只允许操作白名单目录。", "tools": [ { "type": "mcp", "transport": "sse", "url": "http://localhost:3000/sse", "name": "filesystem" } ] }, "litellm": { "api_base": "https://taotoken.net/api", "api_key": "${TAOTOKEN_API_KEY}", "drop_params": true } }这里有个关键点:drop_params: true必须开。因为不同模型对 Function Calling 的参数支持不一致,LiteLLM 会自动丢弃不支持的字段,避免报 400。TaoToken 的 API 通道兼容 OpenAI 格式,所以api_base直接填 https://taotoken.net/api 就行,不需要加/v1后缀。
注意:allowed_roots 白名单是企业级 MCP 中间层的核心安全边界。没有这个配置,Agent 理论上可以读写任意路径,这是生产环境绝对不能接受的。
4. FastMCP Server 实现与 ADK Client 接入
配置就绪后,开始写代码。整个项目结构如下:
mcp-agent-demo/ ├── config.toml ├── settings.json ├── filesystem_server.py ├── agent.py └── requirements.txt先装依赖:
pip install fastmcp google-adk litellm4.1 FastMCP Server:封装文件系统访问
filesystem_server.py 的核心是把文件操作暴露成 MCP 工具,同时加上白名单校验:
# filesystem_server.py import os import tomllib from pathlib import Path from fastmcp import FastMCP # 读取配置 with open("config.toml", "rb") as f: cfg = tomllib.load(f) ALLOWED_ROOTS = [Path(p).resolve() for p in cfg["mcp_server"]["security"]["allowed_roots"]] mcp = FastMCP(cfg["mcp_server"]["name"]) def _check_path(target: str) -> Path: """校验路径是否在白名单内,防止越权访问""" p = Path(target).resolve() for root in ALLOWED_ROOTS: if root == p or root in p.parents: return p raise PermissionError(f"路径 {p} 不在允许的白名单目录内") @mcp.tool() def list_directory(path: str) -> list[str]: """列出指定目录下的文件和子目录""" safe = _check_path(path) if not safe.is_dir(): raise NotADirectoryError(f"{safe} 不是目录") return sorted([item.name for item in safe.iterdir()]) @mcp.tool() def read_file(path: str, max_bytes: int = 4096) -> str: """读取文件内容,默认最多读取 4096 字节""" safe = _check_path(path) if not safe.is_file(): raise FileNotFoundError(f"{safe} 不存在") with open(safe, "r", encoding="utf-8", errors="ignore") as f: return f.read(max_bytes) @mcp.tool() def write_file(path: str, content: str) -> str: """写入文件内容,仅允许白名单目录""" safe = _check_path(path) safe.parent.mkdir(parents=True, exist_ok=True) with open(safe, "w", encoding="utf-8") as f: f.write(content) return f"已写入 {len(content)} 字符到 {safe}" if __name__ == "__main__": mcp.run( transport=cfg["mcp_server"]["transport"], host=cfg["mcp_server"]["host"], port=cfg["mcp_server"]["port"], )启动 Server:
python filesystem_server.py看到Uvicorn running on http://0.0.0.0:3000就说明 MCP Server 起来了。这里用 SSE 传输,因为 ADK 的 MCPToolset 对 SSE 支持最稳定。
4.2 ADK Agent:作为 MCP Client 调用
agent.py 里构建 Agent,通过 MCPToolset 连接上面的 Server:
# agent.py import os import json from google.adk.agents import Agent from google.adk.tools.mcp_tool import MCPToolset, SseServerParams with open("settings.json", "r", encoding="utf-8") as f: settings = json.load(f) # 从环境变量注入 TaoToken Key os.environ["OPENAI_API_KEY"] = os.environ["TAOTOKEN_API_KEY"] mcp_toolset = MCPToolset( connection_params=SseServerParams( url=settings["agent"]["tools"][0]["url"] ) ) root_agent = Agent( name=settings["agent"]["name"], model=settings["agent"]["model"], instruction=settings["agent"]["instruction"], tools=[mcp_toolset], )启动 ADK Web UI:
adk web --port 8000浏览器打开 http://localhost:8000 ,选择 file_agent,输入「列出 /data/workspace 下的所有文件」,你会看到 Agent 先做一次 LLM 调用筛选出list_directory工具,然后通过 MCP 发起实际调用,最后再调一次 LLM 组装自然语言回复。整个链路里,Agent 完全不知道文件系统是怎么实现的,它只认 MCP 协议。
5. 验证请求与成功结果
配置和代码都就位后,做一次端到端验证。我习惯分三层验证,从下往上排查。
第一层,直接测 MCP Server。用 curl 发一个 SSE 连接请求:
curl -N http://localhost:3000/sse正常会返回event: endpoint和data: /messages/?session_id=xxx。如果连不上,说明 Server 没起来或端口被占。
第二层,用 MCP Inspector 可视化验证。这是官方调试工具:
npx @modelcontextprotocol/inspector浏览器打开后,Transport Type 选 SSE,URL 填http://localhost:3000/sse,点 Connect。然后在 Tools 页面点 List Tools,应该能看到list_directory、read_file、write_file三个工具。点开list_directory,参数填/data/workspace,执行后返回文件列表,说明工具层没问题。
第三层,走完整 Agent 链路。在 ADK Web UI 里输入:
帮我看看 /data/workspace 下有哪些文件,然后读取第一个文件的前 100 个字符预期结果:Agent 会连续调用list_directory和read_file两个 MCP 工具,最后用自然语言汇总。如果这一步成功,说明 ADK + MCP + TaoToken 三层全部打通。
实测下来,从输入到返回大概 3-5 秒,取决于模型响应速度。TaoToken 的 API 通道在这里的作用是统一了 LLM 调用入口,你换模型只需要改 config.toml 里的 model 字段,MCP 层完全不用动。
6. 本篇常见错排查
搭这套中间层,我踩过的坑主要集中在几个地方,列出来帮你省时间。
报错一:MCPToolset connection refused
原因通常是 MCP Server 没启动,或者 host 配成了127.0.0.1而 ADK 在容器里跑。解决:确认python filesystem_server.py在运行,host 改成0.0.0.0,ADK 侧 URL 用http://host.docker.internal:3000/sse(容器场景)。
报错二:PermissionError: 路径不在白名单
这是安全校验生效了,不是 bug。检查 config.toml 里的allowed_roots是否包含你操作的目录。注意路径要写绝对路径,~不会被自动展开。
报错三:400 Bad Request - unsupported parameter
LiteLLM 转发时带了模型不支持的参数。解决:settings.json 里确保drop_params: true,同时把temperature、max_tokens这些放在 config.toml 的[llm]段,让 LiteLLM 统一处理。
报错四:Agent 不调用工具,直接瞎编答案
模型不支持 Function Calling,或者 instruction 没写清楚。解决:去 https://taotoken.net/models 确认模型支持工具调用,instruction 里明确写「必须通过 MCP 工具访问文件系统,不要凭记忆回答」。
报错五:SSE 连接频繁断开
企业级场景下长连接容易被中间设备掐断。解决:在 MCP Server 侧加心跳,FastMCP 支持ping_interval参数;或者改用 stdio 传输,适合单机部署。
排查顺序建议从下往上:先 curl 测 Server,再 Inspector 测工具,最后 ADK 测 Agent。哪一层断了就修哪一层,不要跳着查。
如果你在接入过程中遇到 Key 或通道问题,可以直接去 https://taotoken.net/api-keys 重新生成一个,然后在 https://taotoken.net/doc 对照接入文档检查 base_url 和 header 格式。长期做编码类 Agent 的话,Coding Plan 的额度模型更适合高频工具调用场景,可以去 https://taotoken.net/coding-plan 看看配额策略。整套中间层骨架跑通后,你只需要往 FastMCP 里加新工具,Agent 侧零改动,这就是 MCP 中间层在企业级架构里的真正价值。