尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

MCP 协议完全指南:从 JSON-RPC 原理到 LangGraph 集成,打造即插即用的 AI Agent 工具生态

MCP 协议完全指南:从 JSON-RPC 原理到 LangGraph 集成,打造即插即用的 AI Agent 工具生态 1. 为什么你的 LangGraph Agent 总在重复造轮子如果你正在用 LangGraph 搭 AI Agent大概率写过这样的代码把查天气、读数据库、调内部 API 的函数一个个用tool装饰器包起来塞进tools列表然后祈祷模型能正确调用。项目小的时候没问题一旦工具超过十个、需要跨团队复用、或者想让 Node.js 写的服务也能被 Agent 调用这套做法就开始崩了。问题出在“N×M 集成”上。假设你有 5 个 AI 应用、8 个外部工具理论上要写 40 份连接代码。每换一个模型框架工具层就得重写一遍。MCPModel Context Protocol模型上下文协议就是冲着这个痛点来的——它把工具从 Agent 代码里彻底剥离变成一个独立进程通过标准协议通信。你可以把它理解成 AI 世界的 USB-C 接口任何符合 MCP 规范的工具都能插进任何支持 MCP 的宿主应用。这篇文章面向正在构建 AI Agent 工具生态的开发者从 JSON-RPC 通信原理讲到 LangGraph 集成落地。我会给出可复制的 FastMCP Server 代码、MultiServerMCPClient配置、手动管理 Session 的写法以及本地验证工具即插即用的完整步骤。读完你能自己搭一个 MCP Server并让 LangGraph Agent 像调用本地函数一样调用它。先说清楚 MCP 的三个角色不然后面配置容易懵。Host 是你的 AI 应用本体比如 LangGraph Agent它负责决策和调度Client 是 Host 为每个 Server 创建的通信组件维护专用连接Server 是提供具体能力的独立程序只做一件事。关键设计原则是Server 无法读取完整对话历史也看不到其他 Server它只接收当前请求的参数。这个隔离性既是安全边界也是可组合性的基础。MCP 协议分两层。数据层基于 JSON-RPC 2.0定义了生命周期管理握手、初始化、关闭、核心原语Tools、Resources、Prompts和通知机制。传输层目前有两种标准机制STDIO 和 HTTPSSE。STDIO 把 Server 作为子进程启动通过 stdin/stdout 交换 JSON-RPC 消息适合本地 CLI 和桌面应用HTTPSSE 让 Server 独立运行通过 SSE 接收消息、POST 发送消息适合远程部署。需要提醒的是SSE 传输在 2025 年 3 月已被标记为弃用新项目建议用 Streamable HTTP。理解了这些你就能明白为什么 MCP Server 写起来那么简单——复杂编排全在 Host 侧Server 只管执行。接下来进入实操。2. TaoToken 前置给 Agent 一个稳定的模型出口在写 MCP Server 之前得先解决模型调用的问题。LangGraph Agent 的“大脑”是 LLM如果模型接口不稳定工具调用链路再优雅也跑不起来。我自己的做法是把模型出口统一到 TaoToken 上它的 API 兼容 OpenAI 格式ChatOpenAI直接改base_url就能用省去在多个供应商之间切换的麻烦。TaoToken 在这里扮演的是模型网关角色不是 MCP 的一部分但它决定了你的 Agent 能不能稳定地做工具决策。MCP 负责“手脚”TaoToken 负责“大脑”的输入输出通道两者配合才能跑通完整链路。你需要先拿到 API Key。访问控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。创建后复制那串sk-开头的字符串后面配置里要用。模型选择上工具调用场景建议用指令遵循能力强的模型。你可以在模型对话页面先测一下模型对 JSON 格式工具调用的响应质量https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果只是本地验证 MCP 链路用便宜的小模型就够了如果要跑复杂的多工具编排选推理能力强的。环境变量配置建议单独放.env文件别硬编码在代码里# .env OPENAI_API_KEYsk-你的TaoToken密钥 OPENAI_BASE_URLhttps://taotoken.net/api注意OPENAI_BASE_URL后面不加 UTM 参数API 地址就是https://taotoken.net/api。这个地址是 OpenAI 兼容端点langchain-openai的ChatOpenAI会自动读取OPENAI_API_KEY和OPENAI_BASE_URL环境变量所以代码里不用显式传参。如果你打算长期跑编码类 Agent或者需要频繁调用工具做多步推理可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。它针对高频编码场景做了优化比按量计费更适合持续开发。配置好模型出口后先跑一个最小验证确认ChatOpenAI能正常返回# test_model.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() model ChatOpenAI(modelgpt-4o-mini) resp model.invoke(回复两个字收到) print(resp.content)如果输出“收到”说明模型通道没问题可以进入 MCP Server 的编写。如果报 401检查 Key 是否复制完整如果报连接错误检查OPENAI_BASE_URL是否写成了https://taotoken.net/api不要带路径后缀。这一步看起来简单但很多人在后面调试 MCP 工具调用失败时其实是模型通道本身就不通白白浪费排查时间。先把模型跑通再叠 MCP。3. 可复制配置FastMCP Server 与 LangGraph 接入这一节是全文的核心我会给出三个可直接复制的配置片段FastMCP Server 本体、MultiServerMCPClient的 JSON 配置、以及 LangGraph Agent 的接入代码。路径和参数都按实际可运行的标准写你改一下文件路径就能跑。先装依赖pip install fastmcp langchain-mcp-adapters langgraph langchain-openai python-dotenv3.1 FastMCP Servermath_server.pyFastMCP 是 MCP 官方 Python SDK 的高层封装用装饰器就能把普通函数变成 MCP 工具。它自动读取类型提示和文档字符串生成 JSON Schema模型靠这个 Schema 理解工具用途。# math_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(Math) mcp.tool() def add(a: int, b: int) - int: Add two numbers return a b mcp.tool() def multiply(a: int, b: int) - int: Multiply two numbers return a * b mcp.tool() def subtract(a: int, b: int) - int: Subtract two numbers return a - b if __name__ __main__: mcp.run(transportstdio)这里mcp.tool()是关键它把函数注册为 MCP 工具。文档字符串会被 AI 读取所以别写“TODO”之类的占位。mcp.run(transportstdio)让 Server 监听标准输入输出作为子进程被 Client 启动。3.2 MultiServerMCPClient 配置langchain-mcp-adapters提供了MultiServerMCPClient用一个字典配置多个 Server。这个字典就是你的 MCP 连接配置等价于 JSON 配置# mcp_config.py MCP_SERVERS { math: { transport: stdio, command: python, args: [/absolute/path/to/math_server.py], }, # 可以继续加更多 Server # weather: { # transport: http, # url: http://localhost:8000/mcp, # }, }注意args里必须用绝对路径相对路径在子进程启动时会找不到文件。这是最常见的坑之一。3.3 LangGraph Agent 接入代码# agent.py import asyncio from dotenv import load_dotenv from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.agents import create_agent from langchain_openai import ChatOpenAI from mcp_config import MCP_SERVERS load_dotenv() async def main(): client MultiServerMCPClient(MCP_SERVERS) tools await client.get_tools() model ChatOpenAI(modelgpt-4o-mini) agent create_agent(model, tools) response await agent.ainvoke( {messages: [{role: user, content: whats (3 5) x 12?}]} ) print(response[messages][-1].content) if __name__ __main__: asyncio.run(main())client.get_tools()是核心方法它会与所有配置的 Server 建立连接拉取工具列表并自动转换成 LangChain 兼容的 Tool 对象。create_agent内部构建了标准 ReAct 图Agent 调用工具时完全感知不到背后是 MCP。如果你需要更精细的控制比如自定义认证或会话管理可以手动管理 Session# agent_manual.py import asyncio from dotenv import load_dotenv from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain.agents import create_agent from langchain_openai import ChatOpenAI load_dotenv() async def main(): server_params StdioServerParameters( commandpython, args[/absolute/path/to/math_server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) model ChatOpenAI(modelgpt-4o-mini) agent create_agent(model, tools) response await agent.ainvoke( {messages: [{role: user, content: whats 10 - 3?}]} ) print(response[messages][-1].content) if __name__ __main__: asyncio.run(main())await session.initialize()是 MCP 协议要求的握手Client 和 Server 在此交换能力和协议版本信息。少了这一步后续load_mcp_tools会失败。项目结构建议这样组织my-mcp-agent-project/ ├── .env ├── math_server.py ├── mcp_config.py └── agent.py配置片段都齐了下一节验证实际运行结果。4. 验证请求从用户提问到工具返回的完整链路配置写完后跑起来看结果。执行python agent.py如果一切正常你会看到类似The result is 96的输出。但光看最终结果不够我们得确认 MCP 工具真的被调用了而不是模型自己算出来的。先单独验证 MCP Server 能启动。直接运行python math_server.py如果没有任何输出且进程挂起说明 Server 在等待 stdin 输入这是正常的——STDIO 传输模式下它作为子进程运行。按 CtrlC 退出。更可靠的验证方式是写一个最小 Client 测试脚本# test_mcp.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython, args[/absolute/path/to/math_server.py], ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具) for t in tools.tools: print(f - {t.name}: {t.description}) result await session.call_tool(add, {a: 3, b: 5}) print(fadd(3,5) {result.content[0].text}) if __name__ __main__: asyncio.run(main())运行后应该输出可用工具 - add: Add two numbers - multiply: Multiply two numbers - subtract: Subtract two numbers add(3,5) 8这说明 MCP Server 正常暴露工具且能通过 JSON-RPC 调用。这一步过了再跑 Agent 就有底了。回到 Agent 执行流程以(35)×12为例完整链路是这样的用户输入进入 LangGraph AgentAgent 内部 messages 列表追加 HumanMessageLLM 第一次思考看到可用工具列表[add, multiply, subtract]推理出需要先算 35 再乘 12生成两个 tool_calls路由判断检测到 tool_calls跳转到 ToolNodeToolNode 遍历调用列表对每个 tool_call 识别工具名调用 MCP Client 的call_tool方法MCP Client 通过 STDIO 向 Server 发送 JSON-RPC 请求Server 执行函数返回结果Client 把结果返回给 ToolNode包装成 ToolMessage 追加到 messages流程回到 chatbot 节点LLM 第二次思考看到工具结果后生成纯文本回复路由再次判断无 tool_calls返回 END。想看到中间过程可以在 Agent 调用后打印完整 messagesfor msg in response[messages]: print(f[{msg.__class__.__name__}] {msg.content}) if hasattr(msg, tool_calls) and msg.tool_calls: print(f tool_calls: {msg.tool_calls})你会看到 AIMessage 带 tool_calls、ToolMessage 带结果、最后 AIMessage 带文本回复。这个链路确认后说明 MCP 工具已经即插即用地接入了 LangGraph。STDIO 传输的关键点Client 把 Server 作为子进程启动消息通过 stdin/stdout 传输每条消息以换行符分隔且必须是有效 JSON-RPC 格式。Server 的日志要写 stderr写 stdout 会污染协议消息导致解析失败——这是新手最容易踩的坑。5. 本篇常见错排查401、local proxy failed 与 reading choices这一节整理实际调试中高频出现的报错每个都给出定位思路和修复方法。401 Unauthorized模型调用返回 401说明 TaoToken 的 Key 有问题。检查.env里OPENAI_API_KEY是否以sk-开头且完整复制OPENAI_BASE_URL是否为https://taotoken.net/api。如果 Key 正确仍报 401去控制台确认密钥是否被禁用或额度耗尽。注意ChatOpenAI默认读环境变量如果你在代码里显式传了api_key参数确保它没被覆盖成空值。local proxy failed / Connection error这个报错通常出现在client.get_tools()阶段说明 MCP Client 无法启动 Server 子进程。最常见原因是args里用了相对路径。改成绝对路径比如/Users/yourname/project/math_server.py。另一个原因是command指定的python不在 PATH 里可以换成sys.executable的绝对路径。如果 Server 启动时抛异常子进程会立即退出Client 侧看到的就是连接失败——单独运行python math_server.py确认 Server 本身能启动。Error reading choices / 响应解析失败这个报错来自模型侧通常是base_url配置错误导致返回了非 OpenAI 格式的响应。确认OPENAI_BASE_URL是https://taotoken.net/api不要多加/v1或/chat/completions后缀langchain-openai会自动拼接。如果用的是自定义ChatOpenAI实例检查base_url参数是否被其他配置覆盖。OAuth / 认证握手失败如果你接的是需要 OAuth 的远程 MCP Serversession.initialize()可能因为 token 缺失或过期失败。检查 Server 配置里的认证头是否正确传递。本地 STDIO Server 一般不需要 OAuth如果报这个错先确认你没把 HTTP 传输的配置误用到 STDIO 上。工具列表为空client.get_tools()返回空列表说明 Client 连上了 Server 但没拉到工具。检查mcp.tool()装饰器是否加在函数上函数是否有类型提示和文档字符串。FastMCP 靠这些生成 Schema缺了会导致工具注册失败。另外确认mcp.run(transportstdio)在if __name__ __main__:块里否则子进程启动时可能不执行。CC Switch / Cline MCP / Codex auth.json 三件套如果你在 Cline 或类似工具里配置 MCP需要同时填 Base URL、Key、Model ID 三项。Base URL 填https://taotoken.net/apiKey 填sk-开头的密钥Model ID 填你选的模型名如gpt-4o-mini。缺任何一项都会导致工具调用链路断裂。Codex 的auth.json里对应字段是api_base、api_key、model路径通常在~/.codex/auth.json。排查顺序建议先确认模型通道通跑test_model.py再确认 MCP Server 能独立启动跑test_mcp.py最后跑 Agent。分层排查能快速定位问题在哪一层。6. 把工具生态跑起来从单 Server 到多 Server 组合单 Server 跑通后MCP 的真正价值在于组合。你可以在MCP_SERVERS字典里加更多 ServerMultiServerMCPClient会自动聚合所有工具。比如加一个天气 ServerMCP_SERVERS { math: { transport: stdio, command: python, args: [/absolute/path/to/math_server.py], }, weather: { transport: stdio, command: python, args: [/absolute/path/to/weather_server.py], }, }Agent 侧代码完全不用改client.get_tools()会把两个 Server 的工具合并成一个列表。模型看到的是统一的工具集调用时 Client 自动路由到对应的 Server。这就是“即插即用”的实际含义——加工具不用改 Agent 代码。如果你要把 MCP 工具接入 Claude Code 这类编码 Agent配置方式类似在 MCP 配置文件里填 Server 启动命令即可。接入文档里有各客户端的详细配置示例https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。API Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 需要新建或轮换密钥时去那里操作。最后说一个实际经验MCP Server 的粒度要控制好。一个 Server 只做一类事比如数学计算一个、文件操作一个、数据库查询一个。别把所有工具塞进一个 Server那样既失去了隔离性也让复用变难。Server 之间通过 Host 编排组合这才是 MCP 架构的设计意图。跑通这条链路后你的 LangGraph Agent 工具系统就从“内置函数”变成了“可插拔外设”。下次要加新能力写个独立 Server 注册进去就行Agent 代码一行不动。
返回列表