[Ai Agent] 11 MCP进阶：手写客户端，让MCP连接万物（Client）——TaoToken 统一 Key 接入 Stdio 与 Streamable HTTP

1. 为什么我要手写 MCP Client而不是一直用 GUI上一章我们用 FastMCP 造了个天气服务CherryStudio 点几下就连上了看起来一切都很美好。但真实工程里Agent 往往跑在 Linux 服务器上的 Python 脚本里没有图形界面或者需要同时连接本地的高德地图和远程的天气服务还要统一调度。这时候 GUI 就不够用了必须自己写 MCP Client。MCPModel Context Protocol能做什么简单说它让智能体用统一协议调用任意语言写的工具。适合谁适合已经会写 Python、想让 Agent 真正接入外部工具的人。我试过用封装库快速搭起来但只有亲手写过连接管理和协议适配你才会真正理解uvx到底怎么启动子进程JSON-RPC 消息怎么在管道里流转Streamable HTTP 到底比 SSE 强在哪这篇就带你从零手写一个支持 Stdio 与 Streamable HTTP 双模传输的 MCP Client并用 TaoToken 统一 Key 接入模型通道最后给出可复制的配置骨架和连通性验证动作。全程代码可跟做踩过的坑我也会标出来。2. TaoToken 前置统一 Key 与 API 通道在写客户端之前先把模型通道准备好。MCP Client 本身只负责连接工具但 Agent 的“大脑”还是 LLM所以我们需要一个稳定的 API 入口。TaoToken 提供统一 Key兼容 OpenAI 风格的接口接入文档在 https://taotoken.net/api 官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。你需要先拿到 API Key然后配置到环境变量里。我习惯用config.py集中管理避免 Key 散落在代码各处# config.py import os OPENAI_API_KEY os.getenv(TAOTOKEN_API_KEY, sk-你的Key) OPENAI_BASE_URL https://taotoken.net/api AMAP_MAPS_API_KEY os.getenv(AMAP_MAPS_API_KEY, 你的高德Key)注意不要把 Key 硬编码进提交到 Git 的文件里用环境变量或.env加载。TaoToken 的 Key 管理页面在 https://taotoken.net/api-keys 可以随时轮换。如果你只是验证模型对话是否通可以直接用模型对话页面 https://taotoken.net/models 试一条消息如果打算长期跑编码类 Agent建议看 Coding Plan https://taotoken.net/coding-plan 额度更划算。3. 可复制配置Stdio 与 Streamable HTTP 双模骨架MCP 定义了多种 Transport当前主流是 Stdio 和 Streamable HTTPSSE 已逐步淘汰。Stdio 面向本地开发本质是父子进程间的匿名管道通信Streamable HTTP 面向远程协作把工具变成可远程调用的 Web 服务。3.1 Stdio 模式配置Stdio 的启动靠npxNode.js 生态或uvxPython 生态临时执行即用即走不污染全局环境。下面是一个settings.json风格的配置骨架很多 MCP 宿主都认这个格式{ mcpServers: { 高德地图: { transport: stdio, command: npx, args: [-y, amap/amap-maps-mcp-server], env: { AMAP_MAPS_API_KEY: 你的高德Key } }, 本地天气: { transport: stdio, command: python, args: [-m, m10_mcp_basics.stdio_server], env: null } } }如果你更喜欢 TOMLconfig.toml可以这样写[[mcp_servers]] name 高德地图 transport stdio command npx args [-y, amap/amap-maps-mcp-server] [mcp_servers.env] AMAP_MAPS_API_KEY 你的高德Key3.2 Streamable HTTP 模式配置HTTP 模式下MCP Server 必须提前启动并长期运行客户端只消费服务不管理生命周期。配置里换成url即可{ mcpServers: { 高德地图远程: { transport: streamable_http, url: https://mcp.amap.com/mcp?key你的高德Key }, 本地天气服务: { transport: streamable_http, url: http://127.0.0.1:8001/mcp } } }注意测试本地 HTTP 服务前必须先运行对应的服务器文件比如streamable_http_server.py否则连接会直接失败。3.3 用适配器快速加载工具有了配置用langchain-mcp-adapters可以快速把两种传输统一成工具列表import os from langchain_mcp_adapters.client import MultiServerMCPClient from config import AMAP_MAPS_API_KEY MCP_SERVERS { 高德地图: { transport: stdio, command: npx, args: [-y, amap/amap-maps-mcp-server], env: {**os.environ, AMAP_MAPS_API_KEY: AMAP_MAPS_API_KEY} }, 高德地图远程: { transport: streamable_http, url: fhttps://mcp.amap.com/mcp?key{AMAP_MAPS_API_KEY} } } async def load_tools(): client MultiServerMCPClient(MCP_SERVERS) tools await client.get_tools() print(已加载工具:, [t.name for t in tools]) return toolsget_tools()会向所有注册的 MCP 服务发送list_tools请求Stdio 服务启动子进程读工具声明HTTP 服务发 POST 到/mcp拿元数据最终返回统一的 Tool 列表。4. 验证请求从 JSON-RPC 到成功结果配置写好了怎么确认真的连通最直接的办法是看 JSON-RPC 消息。Stdio 下请求写进 stdin响应从 stdout 读出来{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: search_poi, arguments: {keyword: 西湖} } }响应长这样{ jsonrpc: 2.0, id: 1, result: { content: [{type: text, text: 找到西湖附近 10 个 POI}] } }HTTP 模式下响应格式要自动识别。自研 FastMCP 服务通常返回text/event-stream第三方服务可能直接返回application/json。所以客户端不能盲目response.json()import json import httpx async def call_http_mcp(url, payload): async with httpx.AsyncClient() as client: resp await client.post(url, jsonpayload) content_type resp.headers.get(content-type, ) if text/event-stream in content_type: for line in resp.text.splitlines(): if line.startswith(data:): return json.loads(line[5:]) return resp.json()跑通后控制台会打印出工具名和调用结果。我实测下来四种组合Stdio/HTTP × 自研/高德全部能通说明这套骨架的通用性没问题。5. 本篇常见错排查报错一npx找不到或子进程启动失败。先确认 Node.js 已安装npx --version能输出版本。如果是在无 Node 环境的服务器上改用uvx跑 Python 版 MCP Server或者直接走 Streamable HTTP 模式。报错二HTTP 连接返回 404 或连接被拒。检查服务是否已启动端口是否对得上。本地服务用127.0.0.1而不是localhost避免 DNS 解析问题。远程服务确认 URL 里的key参数没写错。报错三工具列表为空。多半是env没传对比如高德的AMAP_MAPS_API_KEY没注入。Stdio 模式下子进程继承的是你显式传入的env不是父进程的全部环境变量所以要用{**os.environ, AMAP_MAPS_API_KEY: ...}合并。报错四程序退出后留下僵尸进程。这是没管好子进程生命周期。用AsyncExitStack把子进程、管道、会话都注册进去退出时按后进先出顺序自动清理from contextlib import AsyncExitStack class StdioMCPTransport: def __init__(self): self.exit_stack AsyncExitStack() async def connect(self): transport await self.exit_stack.enter_async_context(stdio_client(self.params)) self.session await self.exit_stack.enter_async_context( ClientSession(transport[0], transport[1]) ) async def cleanup(self): await self.exit_stack.aclose()报错五多工具时所有调用都指向最后一个工具。这是 Python 闭包延迟绑定的经典坑。循环里创建函数时用默认参数立即锁定变量值async def _dynamic_tool_func(tool_nametool_info[name], **kwargs): return await self.client.call_tool(tool_name, kwargs)报错六流式输出一坨一坨蹦。print默认有缓存加flushTrue才能一个字一个字出来。配合astream_events(versionv2)监听on_chat_model_stream和on_tool_start就能看到 Agent 思考、调工具、再总结的全过程。6. 下一步把 Key 和客户端接起来到这里你已经有了一个能同时连 Stdio 和 Streamable HTTP 的 MCP Client 骨架。接下来要做的是把模型通道接上让 Agent 真正跑起来。模型侧用 TaoToken 统一 Key接入文档在 https://taotoken.net/api Key 在 https://taotoken.net/api-keys 管理。如果你要长期跑编码或 Agent 任务Coding Plan https://taotoken.net/coding-plan 会更省心只想先验证模型对话直接去 https://taotoken.net/models 发一条消息就行。把config.py里的OPENAI_BASE_URL指向 TaoTokenOPENAI_API_KEY填你的 Key然后运行主程序async def main(): async with AsyncExitStack() as stack: tools await LangChainMCPAdapter.load_mcp_tools(stack, MCP_SERVER_CONFIGS) app build_graph(available_toolstools) await run_agent_with_streaming(app, 帮我查一下杭州西湖附近的酒店) if __name__ __main__: asyncio.run(main())控制台会依次打印工具加载、工具调用、流式回复。改MCP_SERVER_CONFIGS就能接入数据库、文件系统、地图服务Agent 的能力边界由你决定。