企业官网建设流程全解析

1. 为什么我要手写 MCP Client，而不是一直用 GUI

上一章我们用 FastMCP 造了个天气服务，CherryStudio 点几下就连上了，看起来一切都很美好。但真实工程里，Agent 往往跑在 Linux 服务器上的 Python 脚本里，没有图形界面；或者需要同时连接本地的高德地图和远程的天气服务，还要统一调度。这时候 GUI 就不够用了，必须自己写 MCP Client。

MCP（Model 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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 HTTP（SSE 已逐步淘汰）。Stdio 面向本地开发，本质是父子进程间的匿名管道通信；Streamable HTTP 面向远程协作，把工具变成可远程调用的 Web 服务。

3.1 Stdio 模式配置

Stdio 的启动靠npx（Node.js 生态）或uvx（Python 生态）临时执行，即用即走，不污染全局环境。下面是一个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 } } }

如果你更喜欢 TOML，config.toml可以这样写：

[[mcp_servers]] name = "高德地图" transport = "stdio" command = "npx" args = ["-y", "@amap/amap-maps-mcp-server"] [mcp_servers.env] AMAP_MAPS_API_KEY = "你的高德Key"

3.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": f"https://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 tools

get_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, json=payload) 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_name=tool_info["name"], **kwargs): return await self.client.call_tool(tool_name, kwargs)

报错六：流式输出一坨一坨蹦。print默认有缓存，加flush=True才能一个字一个字出来。配合astream_events(version="v2")监听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指向 TaoToken，OPENAI_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_tools=tools) await run_agent_with_streaming(app, "帮我查一下杭州西湖附近的酒店") if __name__ == "__main__": asyncio.run(main())

控制台会依次打印工具加载、工具调用、流式回复。改MCP_SERVER_CONFIGS就能接入数据库、文件系统、地图服务，Agent 的能力边界由你决定。