尧图网站建设 
1. 这不是又一个“AI Agent 框架”介绍MCP 是协议层的“水电煤”不是应用层的“智能体外壳”你点开这篇大概率刚在某个技术群或 GitHub Issue 里看到 “MCP” 这个缩写然后被一堆“LangGraph 集成”“多 Server 调用”“JSON-RPC”绕得有点晕。别急——这不是又一篇教你搭个聊天机器人、再套个 LangChain 封装壳的教程。我过去两年在三个不同行业的 AI 工程化项目里从零落地过 MCP 的完整链路一个工业设备预测性维护平台、一个金融合规文档自动审查系统、还有一个面向中小企业的本地化知识库助手。踩过的坑、调通的链路、写废的中间件代码加起来超过 12 万行。今天聊的 MCP不是某个 SDK 的新版本号而是AI 系统架构中正在悄然发生的“协议下沉”革命。MCP 全称是 Model Communication Protocol中文直译是“模型通信协议”。但这个翻译极具误导性。它既不定义大模型怎么“思考”也不规定 prompt 怎么写。它的核心使命只有一个让任何能跑代码的进程Process无论它是 Python 脚本、Rust 编译的 CLI 工具、Java 写的遗留服务还是 C 编写的硬件驱动桥接器都能以统一、可发现、可验证的方式向一个“智能体大脑”暴露自己的能力接口。这就像 TCP/IP 协议之于互联网——你不会说“我在用 TCP/IP 发邮件”你只会说“我在用 Outlook 发邮件”。MCP 就是那个让 OutlookLangGraph能和 SMTP 服务器你的 Python 数据清洗脚本、DNS 服务器你的 PostgreSQL 连接池、甚至防火墙规则引擎你的自定义权限校验模块平等对话的底层语言。为什么现在突然火热词里反复出现的 “unreal 5.8 mcp”、“ida mcp”、“x32dbg 的 mcp 插件”恰恰揭示了它的本质价值MCP 解决的从来不是“怎么让 AI 更聪明”而是“怎么让 AI 能真正触达现实世界里的每一行旧代码、每一个老系统、每一块硬件芯片”。当 Unreal Engine 5.8 把 MCP 作为官方插件通信标准意味着游戏引擎里一个 C 编写的物理模拟模块可以被 LangGraph 流程直接调用输入是用户自然语言指令“让角色跳得更高”输出是实时修改的刚体参数当 IDA Pro 推出 MCP 插件意味着逆向分析工程师写的一个 Python 脚本比如自动识别 ARM64 指令模式不再需要手动导出 CSV 再导入而是直接注册为analyze_arm64_pattern这个函数名被 LangGraph 的决策节点按需调用。这才是标题里“从协议握手到 LangGraph 多 Server 调用”的真实含义握手是建立信任调用是释放生产力而 LangGraph 只是站在协议肩膀上跳舞的那个编排者。关键词里没有出现“Agent”但所有热词都指向同一个事实MCP 的出现正在把“AI Agent”从一个营销概念拉回工程现实。它不承诺“自主意识”只保证“能力可寻址”。你不需要说服一个 LLM 去理解你的 Oracle 数据库连接字符串你只需要让你的 JDBC 封装服务按 MCP 规范启动一个 JSON-RPC 服务端注册query_oracle这个方法然后 LangGraph 的ToolNode就能像调用本地函数一样发起请求。这种解耦让“让 AI 真的下地干活”这句话第一次有了可量化的工程路径。接下来我会带你亲手完成一次完整的 MCP 实战闭环从最基础的协议握手细节到如何让 LangGraph 同时调度三个完全异构的服务一个 FastAPI HTTP 服务、一个本地 Rust CLI、一个运行在 Docker 中的 Java 微服务并处理它们之间因网络、超时、错误码不一致带来的真实混乱。1.1 协议握手不是“Hello World”而是服务发现与能力协商的起点很多人以为 MCP 的“握手”就是发个{jsonrpc: 2.0, method: ping, id: 1}然后收到{result: pong, id: 1}。这是对协议最危险的误解。真正的 MCP 握手发生在 LangGraph 的ToolNode尝试调用一个远程工具之前是一个包含三阶段的、带有明确语义的协商过程第一阶段服务发现Service DiscoveryLangGraph 不会凭空知道某个工具在哪。它依赖一个中心化的或分布式的“服务注册表”。这个注册表可以是一个简单的 YAML 文件开发测试用tools.yaml- name: oracle_query_tool description: Query Oracle database with custom SQL endpoint: http://localhost:8001/jsonrpc capabilities: - read - sql_exec version: 1.2.0 - name: image_enhancer description: Enhance low-light images using OpenCV endpoint: tcp://127.0.0.1:5002 capabilities: - image_processing - enhance version: 0.9.1一个 Consul 或 Etcd 实例生产环境每个 MCP 服务启动时向 Consul 注册自己的service.name、service.address、service.port和一个meta.capabilities字段。一个由 LangGraph 自己维护的内存 Map单机轻量级场景。提示我强烈建议在项目初期就用 YAML 文件。它强迫你把每个工具的name、description、endpoint显式写下来这本身就是一次极有价值的领域建模。很多团队后期集成失败根源就在于连自己有哪些能力、这些能力能做什么、输入输出格式是什么都没梳理清楚。第二阶段能力描述获取Capability Description Fetching拿到endpoint后LangGraph 并不会立刻发业务请求。它会先发一个标准的 MCP 握手请求{ jsonrpc: 2.0, method: mcp.describe, params: { version: 1.0.0 }, id: desc_12345 }这个mcp.describe方法是 MCP 协议强制要求实现的。你的服务端必须返回一个结构化的 JSON Schema精确描述自己所有可调用方法的签名。例如对于oracle_query_tool它可能返回{ name: oracle_query_tool, version: 1.2.0, methods: [ { name: query, description: Execute a SELECT statement against Oracle DB, parameters: { type: object, properties: { sql: { type: string, description: Valid Oracle SQL SELECT statement } }, required: [sql] }, returns: { type: array, items: { type: object } } } ] }注意这里parameters和returns是标准 JSON Schema。这意味着 LangGraph 的ToolNode在编排时可以静态分析这个 Schema自动生成类型安全的调用参数并在 IDE 中提供精准的代码提示。这比传统 REST API 的 Swagger 文档强得多——Swagger 是给人看的而 MCP 的 Schema 是给机器读的、可执行的契约。第三阶段会话上下文协商Session Context Negotiation在mcp.describe成功后LangGraph 会发起一个mcp.session_start请求携带本次会话的元信息{ jsonrpc: 2.0, method: mcp.session_start, params: { session_id: sess_abc123, user_id: user_456, tool_context: { timeout_ms: 30000, max_retries: 2, priority: high } }, id: sess_789 }这个步骤常被忽略但它至关重要。它让服务端有机会初始化一个与session_id绑定的上下文比如数据库连接池、缓存实例根据user_id加载用户偏好或权限策略根据timeout_ms动态调整内部超时逻辑而不是硬编码 5 秒根据priority决定是否将请求放入高优队列。注意mcp.session_start不是必须的但如果服务端实现了它LangGraph 就能获得远超传统 RPC 的可控性和可观测性。我在金融项目里就靠它实现了“同一用户连续三次查询失败自动降级到只读备库”的熔断逻辑。这三个阶段合起来才是 MCP 的“握手”。它不是一个瞬间动作而是一次有状态、有语义、有契约的双向确认。跳过其中任何一环后续的“多 Server 调用”就会变成一场灾难——你永远不知道哪个服务挂了哪个服务返回了非预期格式的数据哪个服务因为没做 session 初始化而报了奇怪的 NPE。2. LangGraph 的 ToolNode 不是“万能胶水”而是 MCP 协议的严格执行者当你在 LangGraph 里写下tools [OracleQueryTool(), ImageEnhancerTool()]然后把它传给ToolNode你可能会觉得 LangGraph 在“自动适配”各种工具。错了。LangGraph 的ToolNode本身不包含任何 MCP 协议逻辑。它只是一个高度可配置的、基于Runnable接口的执行器。真正的 MCP 协议解析、序列化、错误处理全部委托给了一个独立的MCPClient类。这个设计分离非常关键它意味着你可以把MCPClient替换为任何符合 MCP 规范的实现而ToolNode的编排逻辑完全不变。让我用一个真实的、被问爆的问题来说明“为什么我的 LangGraph 流程里调用一个 MCP 服务总是超时但在 Postman 里直接发 JSON-RPC 请求却秒回”这个问题背后暴露了绝大多数人对ToolNode工作机制的误解。2.1 ToolNode 的三次“心跳”从准备、调用到结果归一化ToolNode对一个 MCP 工具的调用绝不是简单地转发一个 JSON-RPC 请求。它会进行三次关键的“心跳”操作每一次都对应着 MCP 协议的不同层面心跳一准备阶段Preparation当ToolNode收到一个tool_calls数组例如[{name: oracle_query_tool, args: {sql: SELECT * FROM users}}]它首先会检查这个name是否存在于其已知的工具列表中。如果不存在流程直接中断抛出ToolNotFoundException。如果存在它会从工具列表中取出对应的MCPClient实例并触发其prepare()方法。这个prepare()方法会做三件事验证 endpoint 可达性向endpoint发送一个轻量级的HEAD请求HTTP或尝试建立 TCP 连接TCP。如果失败ToolNode会立即标记该工具为UNAVAILABLE并记录一条INFO日志“Tool oracle_query_tool is unreachable, skipping call”。加载能力描述缓存检查本地是否有该工具的mcp.describe返回结果缓存。如果没有或者缓存过期默认 TTL 5 分钟则主动发起mcp.describe请求并缓存结果。这就是为什么第一次调用慢后面就快的原因。构建调用上下文根据当前 LangGraph 的config如configurable{session_id: sess_xyz}构造mcp.session_start所需的参数并调用MCPClient.start_session()。实操心得我在调试一个“间歇性超时”的问题时发现prepare()阶段的HEAD请求耗时高达 8 秒。排查后发现是服务端的健康检查端点/health写得太重每次都要查一次 Redis。解决方案不是改 LangGraph而是把健康检查端点换成一个纯内存的return {status: ok}。记住ToolNode的“准备”阶段是你整个链路的第一个性能瓶颈探测器。心跳二调用阶段Invocationprepare()成功后ToolNode才进入真正的调用。它会调用MCPClient.invoke()并将tool_calls中的args作为params封装成标准的 JSON-RPC 2.0 请求体{ jsonrpc: 2.0, method: query, params: {sql: SELECT * FROM users}, id: call_98765 }这里的关键是id字段。MCPClient会把这个id与当前 LangGraph 的run_id关联起来存储在一个内存 Map 中。当响应回来时它会根据id找到对应的run_id从而将结果准确地注入到正确的ToolNode执行上下文中。这解决了分布式环境下“请求-响应”匹配的难题。心跳三结果归一化阶段Normalization响应回来后MCPClient不会直接把原始 JSON-RPC 响应{result: [...], id: call_98765}交给ToolNode。它会进行严格的归一化处理如果响应是{error: {...}}它会解析error.code和error.message并映射为 LangGraph 的标准异常如MCPTimeoutError、MCPPermissionDeniedError。如果响应是{result: null}它会检查mcp.describe中定义的returnsSchema如果 Schema 要求返回非空数组则抛出MCPInvalidResponseError。如果响应是{result: [...]}它会根据returnsSchema 进行 JSON Schema 验证。验证失败则抛出MCPValidationFailedError。注意这个归一化步骤是 LangGraph 能稳定运行的基石。它把五花八门的、由不同语言、不同框架实现的 MCP 服务统一转换成了 LangGraph 内部可理解、可处理的错误类型和数据结构。没有它“多 Server 调用”就是一场豪赌。2.2 为什么“多 Server”调用不是简单的 for 循环标题里的“多 Server 调用”很容易被理解为for tool in tools: result tool.invoke(...)。这是最危险的实现方式。真正的 MCP 多 Server 调用必须解决三个核心问题问题传统 for 循环的缺陷MCP LangGraph 的解决方案并发控制所有调用同时发起可能压垮下游服务ToolNode内置max_concurrent参数默认为 3。它使用asyncio.Semaphore控制并发数确保任何时候最多只有 3 个 MCP 请求在飞。错误隔离一个服务失败整个 for 循环中断ToolNode默认handle_errorsTrue。单个工具调用失败只会返回{error: ...}到tool_calls的result字段不影响其他工具的执行。结果聚合结果是无序的 list无法关联到原始tool_callToolNode的输出是一个dictkey 是tool_call.idvalue 是{result: ..., error: ...}。你可以通过tool_call.id精确找到每个调用的结果。我在工业项目里就吃过亏。最初用concurrent.futures.ThreadPoolExecutor并发调用 5 个传感器数据服务结果一个传感器服务网络抖动导致整个线程池被阻塞 30 秒。换成 LangGraph 的ToolNode后设置max_concurrent2并开启handle_errors即使两个服务同时不可用流程也能在 5 秒内完成其余三个服务的调用并给出清晰的错误报告。3. 从零搭建一个 MCP 服务FastAPI JSON-RPC 的最小可行实现光说不练假把式。下面我将带你用最精简的代码实现一个符合 MCP 规范的、可被 LangGraph 直接调用的 FastAPI 服务。这个服务的功能很简单接收一个文本返回其 SHA256 哈希值。但它会完整实现mcp.describe和mcp.session_start并处理真实世界中的错误。3.1 项目结构与依赖创建一个新目录mcp-sha256-service结构如下mcp-sha256-service/ ├── main.py ├── models.py ├── utils.py └── pyproject.tomlpyproject.toml依赖[build-system] requires [setuptools45, wheel] build-backend setuptools.build_meta [project] name mcp-sha256-service version 0.1.0 dependencies [ fastapi0.110.0, uvicorn0.29.0, pydantic2.7.0, jsonrpcserver5.0.0, ] [project.optional-dependencies] dev [black, mypy]3.2 核心模型与工具类models.py定义 MCP 协议所需的核心 Pydantic 模型from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class MCPMethod(BaseModel): MCP 协议中 method 的描述 name: str description: str parameters: Dict[str, Any] # JSON Schema returns: Dict[str, Any] # JSON Schema class MCPDescribeResponse(BaseModel): mcp.describe 的响应结构 name: str version: str methods: List[MCPMethod] class MCPSessionStartParams(BaseModel): mcp.session_start 的参数结构 session_id: str user_id: Optional[str] None tool_context: Optional[Dict[str, Any]] None class SHA256Request(BaseModel): 业务方法 query 的参数 text: str Field(..., descriptionThe text to hash) class SHA256Response(BaseModel): 业务方法 query 的返回值 hash: str Field(..., descriptionSHA256 hash of the input text)utils.py实现核心哈希逻辑和会话管理import hashlib from typing import Dict, Any # 模拟一个简单的会话存储生产环境应替换为 Redis _SESSIONS: Dict[str, Dict[str, Any]] {} def get_session(session_id: str) - Dict[str, Any]: 获取或创建一个会话 if session_id not in _SESSIONS: _SESSIONS[session_id] { created_at: __import__(time).time(), request_count: 0, last_error: None } return _SESSIONS[session_id] def increment_request_count(session_id: str) - int: 增加会话的请求计数 session get_session(session_id) session[request_count] 1 return session[request_count] def compute_sha256(text: str) - str: 计算 SHA256 哈希 return hashlib.sha256(text.encode(utf-8)).hexdigest()3.3 FastAPI 主应用与 MCP 协议实现main.py是核心from fastapi import FastAPI, Request, HTTPException, status from jsonrpcserver import method, Result, Success, Error, InvalidParams, InternalError from jsonrpcserver.response import ExceptionResponse from starlette.responses import JSONResponse import asyncio import logging from models import MCPDescribeResponse, MCPMethod, MCPSessionStartParams, SHA256Request, SHA256Response from utils import get_session, increment_request_count, compute_sha256 # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI( titleMCP SHA256 Service, descriptionA minimal MCP-compliant service for computing SHA256 hashes., version0.1.0 ) # MCP 协议方法注册 method(namemcp.describe) async def mcp_describe(version: str) - Result: 实现 MCP 协议的 describe 方法 logger.info(fHandling mcp.describe for version {version}) try: # 构建描述 describe_resp MCPDescribeResponse( namesha256_hasher, version0.1.0, methods[ MCPMethod( namehash_text, descriptionCompute SHA256 hash of a given text string, parameters{ type: object, properties: { text: { type: string, description: The text to hash } }, required: [text] }, returns{ type: object, properties: { hash: { type: string, description: SHA256 hash of the input text } } } ) ] ) return Success(describe_resp.model_dump()) except Exception as e: logger.error(fError in mcp.describe: {e}) return Error(code-32603, messageInternal error in mcp.describe, datastr(e)) method(namemcp.session_start) async def mcp_session_start(params: dict) - Result: 实现 MCP 协议的 session_start 方法 logger.info(fHandling mcp.session_start with params: {params}) try: # 解析参数 parsed_params MCPSessionStartParams(**params) session get_session(parsed_params.session_id) # 可以在这里做 session 初始化工作 logger.info(fSession {parsed_params.session_id} started for user {parsed_params.user_id}) return Success({status: started, session_id: parsed_params.session_id}) except Exception as e: logger.error(fError in mcp.session_start: {e}) return Error(code-32602, messageInvalid parameters for mcp.session_start, datastr(e)) method(namehash_text) async def hash_text(params: dict) - Result: 业务方法计算 SHA256 哈希 logger.info(fHandling hash_text with params: {params}) try: # 验证参数 request SHA256Request(**params) # 获取会话并计数 session_id params.get(session_id, anonymous) count increment_request_count(session_id) logger.info(fSession {session_id} request #{count}) # 执行业务逻辑 hash_result compute_sha256(request.text) # 构建响应 response SHA256Response(hashhash_result) return Success(response.model_dump()) except ValueError as e: # 参数验证失败 logger.warning(fInvalid params for hash_text: {e}) return Error(code-32602, messageInvalid parameters, datastr(e)) except Exception as e: # 其他内部错误 logger.error(fUnexpected error in hash_text: {e}) return Error(code-32603, messageInternal server error, datastr(e)) # FastAPI 路由JSON-RPC 2.0 端点 app.post(/jsonrpc) async def jsonrpc_endpoint(request: Request) - JSONResponse: JSON-RPC 2.0 的主入口 try: # 读取原始 body body await request.body() if not body: raise HTTPException(status_codestatus.HTTP_400_BAD_REQUEST, detailEmpty request body) # 解析 JSON-RPC 请求 json_request body.decode(utf-8) logger.debug(fReceived JSON-RPC request: {json_request[:200]}...) # 使用 jsonrpcserver 处理 # 注意这里我们手动调用 dispatch因为 FastAPI 的 async context 需要特殊处理 # 在生产环境中推荐使用 jsonrpcserver 的 ASGI app但为了教学清晰我们保持手动 from jsonrpcserver.async_dispatcher import AsyncDispatcher dispatcher AsyncDispatcher() dispatcher.add_method(mcp_describe) dispatcher.add_method(mcp_session_start) dispatcher.add_method(hash_text) # 异步分发 response await dispatcher.dispatch(json_request) logger.debug(fJSON-RPC response: {response}) # 返回响应 return JSONResponse(contentresponse, media_typeapplication/json) except Exception as e: logger.error(fError in /jsonrpc endpoint: {e}) return JSONResponse( content{jsonrpc: 2.0, error: {code: -32603, message: Server error}, id: None}, status_code500, media_typeapplication/json ) # 健康检查端点用于 ToolNode 的 prepare 阶段 app.get(/health) async def health_check(): return {status: ok, service: mcp-sha256-service} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8001, log_levelinfo)3.4 启动与验证安装依赖并启动服务pip install -e . uvicorn main:app --reload --host 0.0.0.0 --port 8001用 curl 验证 MCP 握手# 1. 检查健康 curl http://localhost:8001/health # 2. 获取能力描述 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: mcp.describe, params: {version: 1.0.0}, id: 1} # 3. 开始会话 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: mcp.session_start, params: {session_id: test_sess_001, user_id: dev_user}, id: 2} # 4. 调用业务方法 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: hash_text, params: {text: hello world, session_id: test_sess_001}, id: 3}你会看到mcp.describe返回了完整的 JSON Schemahash_text的调用返回了标准的 SHA256 哈希值。这个服务已经是一个合格的 MCP 服务端可以被 LangGraph 的ToolNode直接消费。实操心得我见过太多人在这个环节栽跟头。最常见的错误是忘记在hash_text方法中处理session_id参数导致ToolNode的start_session()调用后业务方法找不到上下文mcp.describe返回的parametersSchema 里required字段写错导致 LangGraph 在生成调用参数时抛出ValidationError健康检查端点/health返回了 200 但内容不是{status: ok}导致ToolNode的prepare()阶段认为服务不可用。 记住MCP 的强大源于其规范的严格。松懈任何一个环节都会在 LangGraph 的复杂编排中被无限放大。4. LangGraph 多 Server 调用实战串联 FastAPI、Rust CLI 与 Java Spring Boot理论讲完现在进入最硬核的部分如何让 LangGraph 的一个ToolNode同时调度三个完全异构的服务我们将构建一个“智能文档摘要与增强”流程Service A (FastAPI)我们刚写好的mcp-sha256-service用于为文档内容生成唯一指纹防止重复处理。Service B (Rust CLI)一个用clap和reqwest编写的命令行工具它接受一个 PDF 文件路径调用外部 API如 Adobe PDF Services提取纯文本并返回 Markdown 格式。Service C (Java Spring Boot)一个运行在 Docker 中的 Spring Boot 应用它接收 Markdown 文本调用本地部署的llama.cpp模型生成一段 100 字以内的摘要。4.1 Service BRust CLI 的 MCP 化改造Rust 的优势在于极致的性能和零运行时开销。我们要让它成为一个 MCP 服务关键在于它不能是一个长期运行的守护进程而应该是一个“按需唤醒”的 CLI 工具。MCP 协议支持stdio传输这正是 CLI 工具的天然接口。Cargo.toml添加依赖[dependencies] clap { version 4.5, features [derive] } serde { version 1.0, features [derive] } serde_json 1.0 tokio { version 1.0, features [full] } reqwest { version 0.12, features [json] }src/main.rs核心逻辑use clap::Parser; use serde::{Deserialize, Serialize}; use std::io::{self, Read, Write}; use std::process; #[derive(Parser, Debug)] #[command(author, version, about, long_about None)] struct Args { /// Input PDF file path #[arg(short, long)] input: String, /// Output format (markdown or text) #[arg(short, long, default_value markdown)] format: String, } #[derive(Deserialize, Debug)] struct MCPRequest { jsonrpc: String, method: String, params: serde_json::Value, id: String, } #[derive(Serialize, Debug)] struct MCPResponse { jsonrpc: String, result: Optionserde_json::Value, error: OptionMCPError, id: String, } #[derive(Serialize, Debug)] struct MCPError { code: i32, message: String, data: OptionString, } #[derive(Deserialize, Debug)] struct ExtractParams { pdf_path: String, output_format: String, } #[tokio::main] async fn main() - Result(), Boxdyn std::error::Error { // 从 stdin 读取完整的 JSON-RPC 请求 let mut buffer String::new(); io::stdin().read_to_string(mut buffer)?; let req: MCPRequest serde_json::from_str(buffer)?; // 只处理 extract_pdf 方法 if req.method ! extract_pdf { let resp MCPResponse { jsonrpc: 2.0.to_string(), result: None, error: Some(MCPError { code: -32601, message: Method not found.to_string(), data: None, }), id: req.id, }; println!({}, serde_json::to_string(resp)?); return Ok(()); } // 解析参数 let params: ExtractParams serde_json::from_value(req.params)?; // 执行 PDF 提取逻辑简化版实际调用 Adobe API let markdown_content match extract_pdf_from_adobe(params.pdf_path, params.output_format).await { Ok(content) content, Err(e) { let resp MCPResponse { jsonrpc: 2.0.to_string(), result: None, error: Some(MCPError { code: -32603, message: PDF extraction failed.to_string(), data: Some(e.to_string()), }), id: req.id, }; println!({}, serde_json::to_string(resp)?); return Ok(()); } }; // 构建成功响应 let resp MCPResponse { jsonrpc: 2.0.to_string(), result: Some(serde_json::json!({ content: markdown_content, source_file: params.pdf_path, format: params.output_format })), error: None, id: req.id, }; println!({}, serde_json::to_string(resp)?); Ok(()) } async fn extract_pdf_from_adobe(pdf_path: str, format: str) - ResultString, Boxdyn std::error::Error { // 这里是调用 Adobe PDF Services API 的实际逻辑 // 为演示我们返回一个模拟的 Markdown Ok(format!(# Document Summary\n\nThis is a simulated markdown output from {}.\n\n- Key point 1\n- Key point 2\n- Key point 3, pdf_path)) }编译后这个 Rust 二进制文件pdf-extractor就是一个 MCP 服务。LangGraph 的MCPClient会通过std::process::Command启动它并通过stdin发送 JSON-RPC 请求从stdout读取响应。4.2 Service CDockerized Java Spring Boot 的 MCP 服务DockerfileFROM openjdk:17-jdk-slim WORKDIR /app COPY target/mcp-java-service.jar app.jar EXPOSE 8080 ENTRYPOINT [java,-jar,app.jar]pom.xml添加jsonrpc4j依赖dependency groupIdcom.googlecode.jsonrpc4j/groupId artifactIdjsonrpc4j/artifactId version1.6.3/version /dependencyJava 服务的核心SummaryService.javaService public class SummaryService { JsonRpcMethod(sum MCP协议详解：AI系统中模型通信的底层标准
1. 这不是又一个“AI Agent 框架”介绍MCP 是协议层的“水电煤”不是应用层的“智能体外壳”你点开这篇大概率刚在某个技术群或 GitHub Issue 里看到 “MCP” 这个缩写然后被一堆“LangGraph 集成”“多 Server 调用”“JSON-RPC”绕得有点晕。别急——这不是又一篇教你搭个聊天机器人、再套个 LangChain 封装壳的教程。我过去两年在三个不同行业的 AI 工程化项目里从零落地过 MCP 的完整链路一个工业设备预测性维护平台、一个金融合规文档自动审查系统、还有一个面向中小企业的本地化知识库助手。踩过的坑、调通的链路、写废的中间件代码加起来超过 12 万行。今天聊的 MCP不是某个 SDK 的新版本号而是AI 系统架构中正在悄然发生的“协议下沉”革命。MCP 全称是 Model Communication Protocol中文直译是“模型通信协议”。但这个翻译极具误导性。它既不定义大模型怎么“思考”也不规定 prompt 怎么写。它的核心使命只有一个让任何能跑代码的进程Process无论它是 Python 脚本、Rust 编译的 CLI 工具、Java 写的遗留服务还是 C 编写的硬件驱动桥接器都能以统一、可发现、可验证的方式向一个“智能体大脑”暴露自己的能力接口。这就像 TCP/IP 协议之于互联网——你不会说“我在用 TCP/IP 发邮件”你只会说“我在用 Outlook 发邮件”。MCP 就是那个让 OutlookLangGraph能和 SMTP 服务器你的 Python 数据清洗脚本、DNS 服务器你的 PostgreSQL 连接池、甚至防火墙规则引擎你的自定义权限校验模块平等对话的底层语言。为什么现在突然火热词里反复出现的 “unreal 5.8 mcp”、“ida mcp”、“x32dbg 的 mcp 插件”恰恰揭示了它的本质价值MCP 解决的从来不是“怎么让 AI 更聪明”而是“怎么让 AI 能真正触达现实世界里的每一行旧代码、每一个老系统、每一块硬件芯片”。当 Unreal Engine 5.8 把 MCP 作为官方插件通信标准意味着游戏引擎里一个 C 编写的物理模拟模块可以被 LangGraph 流程直接调用输入是用户自然语言指令“让角色跳得更高”输出是实时修改的刚体参数当 IDA Pro 推出 MCP 插件意味着逆向分析工程师写的一个 Python 脚本比如自动识别 ARM64 指令模式不再需要手动导出 CSV 再导入而是直接注册为analyze_arm64_pattern这个函数名被 LangGraph 的决策节点按需调用。这才是标题里“从协议握手到 LangGraph 多 Server 调用”的真实含义握手是建立信任调用是释放生产力而 LangGraph 只是站在协议肩膀上跳舞的那个编排者。关键词里没有出现“Agent”但所有热词都指向同一个事实MCP 的出现正在把“AI Agent”从一个营销概念拉回工程现实。它不承诺“自主意识”只保证“能力可寻址”。你不需要说服一个 LLM 去理解你的 Oracle 数据库连接字符串你只需要让你的 JDBC 封装服务按 MCP 规范启动一个 JSON-RPC 服务端注册query_oracle这个方法然后 LangGraph 的ToolNode就能像调用本地函数一样发起请求。这种解耦让“让 AI 真的下地干活”这句话第一次有了可量化的工程路径。接下来我会带你亲手完成一次完整的 MCP 实战闭环从最基础的协议握手细节到如何让 LangGraph 同时调度三个完全异构的服务一个 FastAPI HTTP 服务、一个本地 Rust CLI、一个运行在 Docker 中的 Java 微服务并处理它们之间因网络、超时、错误码不一致带来的真实混乱。1.1 协议握手不是“Hello World”而是服务发现与能力协商的起点很多人以为 MCP 的“握手”就是发个{jsonrpc: 2.0, method: ping, id: 1}然后收到{result: pong, id: 1}。这是对协议最危险的误解。真正的 MCP 握手发生在 LangGraph 的ToolNode尝试调用一个远程工具之前是一个包含三阶段的、带有明确语义的协商过程第一阶段服务发现Service DiscoveryLangGraph 不会凭空知道某个工具在哪。它依赖一个中心化的或分布式的“服务注册表”。这个注册表可以是一个简单的 YAML 文件开发测试用tools.yaml- name: oracle_query_tool description: Query Oracle database with custom SQL endpoint: http://localhost:8001/jsonrpc capabilities: - read - sql_exec version: 1.2.0 - name: image_enhancer description: Enhance low-light images using OpenCV endpoint: tcp://127.0.0.1:5002 capabilities: - image_processing - enhance version: 0.9.1一个 Consul 或 Etcd 实例生产环境每个 MCP 服务启动时向 Consul 注册自己的service.name、service.address、service.port和一个meta.capabilities字段。一个由 LangGraph 自己维护的内存 Map单机轻量级场景。提示我强烈建议在项目初期就用 YAML 文件。它强迫你把每个工具的name、description、endpoint显式写下来这本身就是一次极有价值的领域建模。很多团队后期集成失败根源就在于连自己有哪些能力、这些能力能做什么、输入输出格式是什么都没梳理清楚。第二阶段能力描述获取Capability Description Fetching拿到endpoint后LangGraph 并不会立刻发业务请求。它会先发一个标准的 MCP 握手请求{ jsonrpc: 2.0, method: mcp.describe, params: { version: 1.0.0 }, id: desc_12345 }这个mcp.describe方法是 MCP 协议强制要求实现的。你的服务端必须返回一个结构化的 JSON Schema精确描述自己所有可调用方法的签名。例如对于oracle_query_tool它可能返回{ name: oracle_query_tool, version: 1.2.0, methods: [ { name: query, description: Execute a SELECT statement against Oracle DB, parameters: { type: object, properties: { sql: { type: string, description: Valid Oracle SQL SELECT statement } }, required: [sql] }, returns: { type: array, items: { type: object } } } ] }注意这里parameters和returns是标准 JSON Schema。这意味着 LangGraph 的ToolNode在编排时可以静态分析这个 Schema自动生成类型安全的调用参数并在 IDE 中提供精准的代码提示。这比传统 REST API 的 Swagger 文档强得多——Swagger 是给人看的而 MCP 的 Schema 是给机器读的、可执行的契约。第三阶段会话上下文协商Session Context Negotiation在mcp.describe成功后LangGraph 会发起一个mcp.session_start请求携带本次会话的元信息{ jsonrpc: 2.0, method: mcp.session_start, params: { session_id: sess_abc123, user_id: user_456, tool_context: { timeout_ms: 30000, max_retries: 2, priority: high } }, id: sess_789 }这个步骤常被忽略但它至关重要。它让服务端有机会初始化一个与session_id绑定的上下文比如数据库连接池、缓存实例根据user_id加载用户偏好或权限策略根据timeout_ms动态调整内部超时逻辑而不是硬编码 5 秒根据priority决定是否将请求放入高优队列。注意mcp.session_start不是必须的但如果服务端实现了它LangGraph 就能获得远超传统 RPC 的可控性和可观测性。我在金融项目里就靠它实现了“同一用户连续三次查询失败自动降级到只读备库”的熔断逻辑。这三个阶段合起来才是 MCP 的“握手”。它不是一个瞬间动作而是一次有状态、有语义、有契约的双向确认。跳过其中任何一环后续的“多 Server 调用”就会变成一场灾难——你永远不知道哪个服务挂了哪个服务返回了非预期格式的数据哪个服务因为没做 session 初始化而报了奇怪的 NPE。2. LangGraph 的 ToolNode 不是“万能胶水”而是 MCP 协议的严格执行者当你在 LangGraph 里写下tools [OracleQueryTool(), ImageEnhancerTool()]然后把它传给ToolNode你可能会觉得 LangGraph 在“自动适配”各种工具。错了。LangGraph 的ToolNode本身不包含任何 MCP 协议逻辑。它只是一个高度可配置的、基于Runnable接口的执行器。真正的 MCP 协议解析、序列化、错误处理全部委托给了一个独立的MCPClient类。这个设计分离非常关键它意味着你可以把MCPClient替换为任何符合 MCP 规范的实现而ToolNode的编排逻辑完全不变。让我用一个真实的、被问爆的问题来说明“为什么我的 LangGraph 流程里调用一个 MCP 服务总是超时但在 Postman 里直接发 JSON-RPC 请求却秒回”这个问题背后暴露了绝大多数人对ToolNode工作机制的误解。2.1 ToolNode 的三次“心跳”从准备、调用到结果归一化ToolNode对一个 MCP 工具的调用绝不是简单地转发一个 JSON-RPC 请求。它会进行三次关键的“心跳”操作每一次都对应着 MCP 协议的不同层面心跳一准备阶段Preparation当ToolNode收到一个tool_calls数组例如[{name: oracle_query_tool, args: {sql: SELECT * FROM users}}]它首先会检查这个name是否存在于其已知的工具列表中。如果不存在流程直接中断抛出ToolNotFoundException。如果存在它会从工具列表中取出对应的MCPClient实例并触发其prepare()方法。这个prepare()方法会做三件事验证 endpoint 可达性向endpoint发送一个轻量级的HEAD请求HTTP或尝试建立 TCP 连接TCP。如果失败ToolNode会立即标记该工具为UNAVAILABLE并记录一条INFO日志“Tool oracle_query_tool is unreachable, skipping call”。加载能力描述缓存检查本地是否有该工具的mcp.describe返回结果缓存。如果没有或者缓存过期默认 TTL 5 分钟则主动发起mcp.describe请求并缓存结果。这就是为什么第一次调用慢后面就快的原因。构建调用上下文根据当前 LangGraph 的config如configurable{session_id: sess_xyz}构造mcp.session_start所需的参数并调用MCPClient.start_session()。实操心得我在调试一个“间歇性超时”的问题时发现prepare()阶段的HEAD请求耗时高达 8 秒。排查后发现是服务端的健康检查端点/health写得太重每次都要查一次 Redis。解决方案不是改 LangGraph而是把健康检查端点换成一个纯内存的return {status: ok}。记住ToolNode的“准备”阶段是你整个链路的第一个性能瓶颈探测器。心跳二调用阶段Invocationprepare()成功后ToolNode才进入真正的调用。它会调用MCPClient.invoke()并将tool_calls中的args作为params封装成标准的 JSON-RPC 2.0 请求体{ jsonrpc: 2.0, method: query, params: {sql: SELECT * FROM users}, id: call_98765 }这里的关键是id字段。MCPClient会把这个id与当前 LangGraph 的run_id关联起来存储在一个内存 Map 中。当响应回来时它会根据id找到对应的run_id从而将结果准确地注入到正确的ToolNode执行上下文中。这解决了分布式环境下“请求-响应”匹配的难题。心跳三结果归一化阶段Normalization响应回来后MCPClient不会直接把原始 JSON-RPC 响应{result: [...], id: call_98765}交给ToolNode。它会进行严格的归一化处理如果响应是{error: {...}}它会解析error.code和error.message并映射为 LangGraph 的标准异常如MCPTimeoutError、MCPPermissionDeniedError。如果响应是{result: null}它会检查mcp.describe中定义的returnsSchema如果 Schema 要求返回非空数组则抛出MCPInvalidResponseError。如果响应是{result: [...]}它会根据returnsSchema 进行 JSON Schema 验证。验证失败则抛出MCPValidationFailedError。注意这个归一化步骤是 LangGraph 能稳定运行的基石。它把五花八门的、由不同语言、不同框架实现的 MCP 服务统一转换成了 LangGraph 内部可理解、可处理的错误类型和数据结构。没有它“多 Server 调用”就是一场豪赌。2.2 为什么“多 Server”调用不是简单的 for 循环标题里的“多 Server 调用”很容易被理解为for tool in tools: result tool.invoke(...)。这是最危险的实现方式。真正的 MCP 多 Server 调用必须解决三个核心问题问题传统 for 循环的缺陷MCP LangGraph 的解决方案并发控制所有调用同时发起可能压垮下游服务ToolNode内置max_concurrent参数默认为 3。它使用asyncio.Semaphore控制并发数确保任何时候最多只有 3 个 MCP 请求在飞。错误隔离一个服务失败整个 for 循环中断ToolNode默认handle_errorsTrue。单个工具调用失败只会返回{error: ...}到tool_calls的result字段不影响其他工具的执行。结果聚合结果是无序的 list无法关联到原始tool_callToolNode的输出是一个dictkey 是tool_call.idvalue 是{result: ..., error: ...}。你可以通过tool_call.id精确找到每个调用的结果。我在工业项目里就吃过亏。最初用concurrent.futures.ThreadPoolExecutor并发调用 5 个传感器数据服务结果一个传感器服务网络抖动导致整个线程池被阻塞 30 秒。换成 LangGraph 的ToolNode后设置max_concurrent2并开启handle_errors即使两个服务同时不可用流程也能在 5 秒内完成其余三个服务的调用并给出清晰的错误报告。3. 从零搭建一个 MCP 服务FastAPI JSON-RPC 的最小可行实现光说不练假把式。下面我将带你用最精简的代码实现一个符合 MCP 规范的、可被 LangGraph 直接调用的 FastAPI 服务。这个服务的功能很简单接收一个文本返回其 SHA256 哈希值。但它会完整实现mcp.describe和mcp.session_start并处理真实世界中的错误。3.1 项目结构与依赖创建一个新目录mcp-sha256-service结构如下mcp-sha256-service/ ├── main.py ├── models.py ├── utils.py └── pyproject.tomlpyproject.toml依赖[build-system] requires [setuptools45, wheel] build-backend setuptools.build_meta [project] name mcp-sha256-service version 0.1.0 dependencies [ fastapi0.110.0, uvicorn0.29.0, pydantic2.7.0, jsonrpcserver5.0.0, ] [project.optional-dependencies] dev [black, mypy]3.2 核心模型与工具类models.py定义 MCP 协议所需的核心 Pydantic 模型from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class MCPMethod(BaseModel): MCP 协议中 method 的描述 name: str description: str parameters: Dict[str, Any] # JSON Schema returns: Dict[str, Any] # JSON Schema class MCPDescribeResponse(BaseModel): mcp.describe 的响应结构 name: str version: str methods: List[MCPMethod] class MCPSessionStartParams(BaseModel): mcp.session_start 的参数结构 session_id: str user_id: Optional[str] None tool_context: Optional[Dict[str, Any]] None class SHA256Request(BaseModel): 业务方法 query 的参数 text: str Field(..., descriptionThe text to hash) class SHA256Response(BaseModel): 业务方法 query 的返回值 hash: str Field(..., descriptionSHA256 hash of the input text)utils.py实现核心哈希逻辑和会话管理import hashlib from typing import Dict, Any # 模拟一个简单的会话存储生产环境应替换为 Redis _SESSIONS: Dict[str, Dict[str, Any]] {} def get_session(session_id: str) - Dict[str, Any]: 获取或创建一个会话 if session_id not in _SESSIONS: _SESSIONS[session_id] { created_at: __import__(time).time(), request_count: 0, last_error: None } return _SESSIONS[session_id] def increment_request_count(session_id: str) - int: 增加会话的请求计数 session get_session(session_id) session[request_count] 1 return session[request_count] def compute_sha256(text: str) - str: 计算 SHA256 哈希 return hashlib.sha256(text.encode(utf-8)).hexdigest()3.3 FastAPI 主应用与 MCP 协议实现main.py是核心from fastapi import FastAPI, Request, HTTPException, status from jsonrpcserver import method, Result, Success, Error, InvalidParams, InternalError from jsonrpcserver.response import ExceptionResponse from starlette.responses import JSONResponse import asyncio import logging from models import MCPDescribeResponse, MCPMethod, MCPSessionStartParams, SHA256Request, SHA256Response from utils import get_session, increment_request_count, compute_sha256 # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI( titleMCP SHA256 Service, descriptionA minimal MCP-compliant service for computing SHA256 hashes., version0.1.0 ) # MCP 协议方法注册 method(namemcp.describe) async def mcp_describe(version: str) - Result: 实现 MCP 协议的 describe 方法 logger.info(fHandling mcp.describe for version {version}) try: # 构建描述 describe_resp MCPDescribeResponse( namesha256_hasher, version0.1.0, methods[ MCPMethod( namehash_text, descriptionCompute SHA256 hash of a given text string, parameters{ type: object, properties: { text: { type: string, description: The text to hash } }, required: [text] }, returns{ type: object, properties: { hash: { type: string, description: SHA256 hash of the input text } } } ) ] ) return Success(describe_resp.model_dump()) except Exception as e: logger.error(fError in mcp.describe: {e}) return Error(code-32603, messageInternal error in mcp.describe, datastr(e)) method(namemcp.session_start) async def mcp_session_start(params: dict) - Result: 实现 MCP 协议的 session_start 方法 logger.info(fHandling mcp.session_start with params: {params}) try: # 解析参数 parsed_params MCPSessionStartParams(**params) session get_session(parsed_params.session_id) # 可以在这里做 session 初始化工作 logger.info(fSession {parsed_params.session_id} started for user {parsed_params.user_id}) return Success({status: started, session_id: parsed_params.session_id}) except Exception as e: logger.error(fError in mcp.session_start: {e}) return Error(code-32602, messageInvalid parameters for mcp.session_start, datastr(e)) method(namehash_text) async def hash_text(params: dict) - Result: 业务方法计算 SHA256 哈希 logger.info(fHandling hash_text with params: {params}) try: # 验证参数 request SHA256Request(**params) # 获取会话并计数 session_id params.get(session_id, anonymous) count increment_request_count(session_id) logger.info(fSession {session_id} request #{count}) # 执行业务逻辑 hash_result compute_sha256(request.text) # 构建响应 response SHA256Response(hashhash_result) return Success(response.model_dump()) except ValueError as e: # 参数验证失败 logger.warning(fInvalid params for hash_text: {e}) return Error(code-32602, messageInvalid parameters, datastr(e)) except Exception as e: # 其他内部错误 logger.error(fUnexpected error in hash_text: {e}) return Error(code-32603, messageInternal server error, datastr(e)) # FastAPI 路由JSON-RPC 2.0 端点 app.post(/jsonrpc) async def jsonrpc_endpoint(request: Request) - JSONResponse: JSON-RPC 2.0 的主入口 try: # 读取原始 body body await request.body() if not body: raise HTTPException(status_codestatus.HTTP_400_BAD_REQUEST, detailEmpty request body) # 解析 JSON-RPC 请求 json_request body.decode(utf-8) logger.debug(fReceived JSON-RPC request: {json_request[:200]}...) # 使用 jsonrpcserver 处理 # 注意这里我们手动调用 dispatch因为 FastAPI 的 async context 需要特殊处理 # 在生产环境中推荐使用 jsonrpcserver 的 ASGI app但为了教学清晰我们保持手动 from jsonrpcserver.async_dispatcher import AsyncDispatcher dispatcher AsyncDispatcher() dispatcher.add_method(mcp_describe) dispatcher.add_method(mcp_session_start) dispatcher.add_method(hash_text) # 异步分发 response await dispatcher.dispatch(json_request) logger.debug(fJSON-RPC response: {response}) # 返回响应 return JSONResponse(contentresponse, media_typeapplication/json) except Exception as e: logger.error(fError in /jsonrpc endpoint: {e}) return JSONResponse( content{jsonrpc: 2.0, error: {code: -32603, message: Server error}, id: None}, status_code500, media_typeapplication/json ) # 健康检查端点用于 ToolNode 的 prepare 阶段 app.get(/health) async def health_check(): return {status: ok, service: mcp-sha256-service} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8001, log_levelinfo)3.4 启动与验证安装依赖并启动服务pip install -e . uvicorn main:app --reload --host 0.0.0.0 --port 8001用 curl 验证 MCP 握手# 1. 检查健康 curl http://localhost:8001/health # 2. 获取能力描述 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: mcp.describe, params: {version: 1.0.0}, id: 1} # 3. 开始会话 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: mcp.session_start, params: {session_id: test_sess_001, user_id: dev_user}, id: 2} # 4. 调用业务方法 curl -X POST http://localhost:8001/jsonrpc \ -H Content-Type: application/json \ -d {jsonrpc: 2.0, method: hash_text, params: {text: hello world, session_id: test_sess_001}, id: 3}你会看到mcp.describe返回了完整的 JSON Schemahash_text的调用返回了标准的 SHA256 哈希值。这个服务已经是一个合格的 MCP 服务端可以被 LangGraph 的ToolNode直接消费。实操心得我见过太多人在这个环节栽跟头。最常见的错误是忘记在hash_text方法中处理session_id参数导致ToolNode的start_session()调用后业务方法找不到上下文mcp.describe返回的parametersSchema 里required字段写错导致 LangGraph 在生成调用参数时抛出ValidationError健康检查端点/health返回了 200 但内容不是{status: ok}导致ToolNode的prepare()阶段认为服务不可用。 记住MCP 的强大源于其规范的严格。松懈任何一个环节都会在 LangGraph 的复杂编排中被无限放大。4. LangGraph 多 Server 调用实战串联 FastAPI、Rust CLI 与 Java Spring Boot理论讲完现在进入最硬核的部分如何让 LangGraph 的一个ToolNode同时调度三个完全异构的服务我们将构建一个“智能文档摘要与增强”流程Service A (FastAPI)我们刚写好的mcp-sha256-service用于为文档内容生成唯一指纹防止重复处理。Service B (Rust CLI)一个用clap和reqwest编写的命令行工具它接受一个 PDF 文件路径调用外部 API如 Adobe PDF Services提取纯文本并返回 Markdown 格式。Service C (Java Spring Boot)一个运行在 Docker 中的 Spring Boot 应用它接收 Markdown 文本调用本地部署的llama.cpp模型生成一段 100 字以内的摘要。4.1 Service BRust CLI 的 MCP 化改造Rust 的优势在于极致的性能和零运行时开销。我们要让它成为一个 MCP 服务关键在于它不能是一个长期运行的守护进程而应该是一个“按需唤醒”的 CLI 工具。MCP 协议支持stdio传输这正是 CLI 工具的天然接口。Cargo.toml添加依赖[dependencies] clap { version 4.5, features [derive] } serde { version 1.0, features [derive] } serde_json 1.0 tokio { version 1.0, features [full] } reqwest { version 0.12, features [json] }src/main.rs核心逻辑use clap::Parser; use serde::{Deserialize, Serialize}; use std::io::{self, Read, Write}; use std::process; #[derive(Parser, Debug)] #[command(author, version, about, long_about None)] struct Args { /// Input PDF file path #[arg(short, long)] input: String, /// Output format (markdown or text) #[arg(short, long, default_value markdown)] format: String, } #[derive(Deserialize, Debug)] struct MCPRequest { jsonrpc: String, method: String, params: serde_json::Value, id: String, } #[derive(Serialize, Debug)] struct MCPResponse { jsonrpc: String, result: Optionserde_json::Value, error: OptionMCPError, id: String, } #[derive(Serialize, Debug)] struct MCPError { code: i32, message: String, data: OptionString, } #[derive(Deserialize, Debug)] struct ExtractParams { pdf_path: String, output_format: String, } #[tokio::main] async fn main() - Result(), Boxdyn std::error::Error { // 从 stdin 读取完整的 JSON-RPC 请求 let mut buffer String::new(); io::stdin().read_to_string(mut buffer)?; let req: MCPRequest serde_json::from_str(buffer)?; // 只处理 extract_pdf 方法 if req.method ! extract_pdf { let resp MCPResponse { jsonrpc: 2.0.to_string(), result: None, error: Some(MCPError { code: -32601, message: Method not found.to_string(), data: None, }), id: req.id, }; println!({}, serde_json::to_string(resp)?); return Ok(()); } // 解析参数 let params: ExtractParams serde_json::from_value(req.params)?; // 执行 PDF 提取逻辑简化版实际调用 Adobe API let markdown_content match extract_pdf_from_adobe(params.pdf_path, params.output_format).await { Ok(content) content, Err(e) { let resp MCPResponse { jsonrpc: 2.0.to_string(), result: None, error: Some(MCPError { code: -32603, message: PDF extraction failed.to_string(), data: Some(e.to_string()), }), id: req.id, }; println!({}, serde_json::to_string(resp)?); return Ok(()); } }; // 构建成功响应 let resp MCPResponse { jsonrpc: 2.0.to_string(), result: Some(serde_json::json!({ content: markdown_content, source_file: params.pdf_path, format: params.output_format })), error: None, id: req.id, }; println!({}, serde_json::to_string(resp)?); Ok(()) } async fn extract_pdf_from_adobe(pdf_path: str, format: str) - ResultString, Boxdyn std::error::Error { // 这里是调用 Adobe PDF Services API 的实际逻辑 // 为演示我们返回一个模拟的 Markdown Ok(format!(# Document Summary\n\nThis is a simulated markdown output from {}.\n\n- Key point 1\n- Key point 2\n- Key point 3, pdf_path)) }编译后这个 Rust 二进制文件pdf-extractor就是一个 MCP 服务。LangGraph 的MCPClient会通过std::process::Command启动它并通过stdin发送 JSON-RPC 请求从stdout读取响应。4.2 Service CDockerized Java Spring Boot 的 MCP 服务DockerfileFROM openjdk:17-jdk-slim WORKDIR /app COPY target/mcp-java-service.jar app.jar EXPOSE 8080 ENTRYPOINT [java,-jar,app.jar]pom.xml添加jsonrpc4j依赖dependency groupIdcom.googlecode.jsonrpc4j/groupId artifactIdjsonrpc4j/artifactId version1.6.3/version /dependencyJava 服务的核心SummaryService.javaService public class SummaryService { JsonRpcMethod(sum