企业官网建设流程全解析

1. 这不是又一个“AI Agent 框架科普”，而是真实跑通 MCP 协议握手、LangGraph 多 Server 调用的全链路实操手记

MCP——最近三个月在工程一线高频出现的词，不是某个新出的模型缩写，也不是某家公司的内部代号，而是一个正在快速落地的标准化协议层。它解决的问题非常具体：当你的 AI Agent 不再是单机玩具，而是要像老式工业控制系统那样，让 LangGraph 编排的决策流，能稳定、可验证、可审计地调用 Unreal Engine 的实时渲染服务、Altium Designer 的 PCB 设计引擎、甚至 IDA Pro 的二进制分析模块时，靠硬编码 HTTP 接口或自定义 socket 协议，已经撑不住了。MCP 就是为这种“异构系统间可信协同”而生的。我上个月在给一家做智能硬件设计平台的客户做技术方案时，第一次把 MCP 协议握手和 LangGraph 的多 Server 调用真正跑通在生产环境里，不是 demo，是每天处理 200+ 个 PCB 设计变更请求的真实链路。它不炫技，但极其务实：JSON-RPC 是它的骨架，类型安全是它的神经，而 LangGraph 是它最趁手的“指挥大脑”。如果你正被“Agent 调用外部工具总出错”、“不同团队开发的服务接口风格五花八门”、“调试一次跨服务调用要翻三份文档”这些问题卡住，那这篇内容就是为你写的。它不讲抽象概念，只讲我在 Windows WSL2 + Ubuntu 22.04 环境下，从零配置 MCP Server、完成三次完整握手、在 LangGraph 中定义并调度两个物理隔离的 Server（一个 Python FastAPI，一个 Rust 实现的硬件仿真器）的每一步命令、每个报错原因、以及那些官方文档里绝不会写的“为什么必须这样配”。

2. 内容整体设计与思路拆解：为什么 MCP 不是另一个轮子，而是协议层的“TCP/IP”

2.1 从“能用”到“可靠”的分水岭：MCP 解决的不是功能问题，而是工程信任问题

很多人第一次接触 MCP，会下意识把它和 LangChain Tools 或 LlamaIndex 的 Connector 做类比。这是最大的认知偏差。LangChain Tools 的本质是 Python 函数封装，它假设调用方和被调用方共享同一个 Python 运行时、同一套依赖版本、甚至同一个进程内存空间。这在本地 demo 里很丝滑，但在真实产线中，它意味着：你无法让一个用 Rust 写的嵌入式固件分析服务，被一个用 TypeScript 写的前端 Agent 直接“import”调用；你也无法让 Altium Designer 这种闭源商业软件，去安装你的 Python 包。MCP 的破局点，恰恰在于它主动放弃“同构运行时”这个幻想，转而拥抱“异构系统间通过标准协议通信”这一更古老、也更健壮的范式。它的设计哲学，和 TCP/IP 协议栈一脉相承：IP 层负责寻址和路由，TCP 层负责可靠传输和流控。MCP 则把“能力发现”、“参数校验”、“错误分类”、“流式响应”这些通用能力，全部下沉到协议层，让上层应用（无论是 LangGraph 还是 Unreal Engine 的蓝图节点）只关心“我要做什么”，而不必操心“怎么连上”、“参数对不对”、“断了怎么办”。

提示：MCP 的核心价值，从来不是“让你更快地写一个 API”，而是“让你敢把关键业务逻辑，放心地交给一个你完全不控制的外部服务来执行”。这背后是一整套基于 JSON Schema 的强类型契约，它比 OpenAPI 更进一步——OpenAPI 描述的是“HTTP 请求长什么样”，而 MCP 描述的是“这个能力本身长什么样”，包括输入参数的语义约束（比如pin_number必须是 1-40 的整数）、输出结果的结构化含义（比如voltage_reading的单位是毫伏，精度是小数点后两位），甚至包括调用失败时应该返回哪一类错误码（invalid_parametervshardware_unavailable）。这才是工程级可靠性的基石。

2.2 为什么选 LangGraph 作为 MCP 的“指挥官”？它和 MCP 是天然互补，而非简单集成

LangGraph 的核心优势，在于它把“状态机”这个古老而强大的概念，用 Pythonic 的方式重新包装。一个典型的 LangGraph 图，由State（当前上下文）、Node（执行单元）和Edge（流转规则）构成。当你把一个 MCP Server 封装成一个 LangGraph Node 时，你获得的远不止是“调用一个函数”那么简单。LangGraph 的State机制，天然承载了 MCP 调用所需的上下文：比如，你在调用一个 PCB 设计 Server 前，State里可能已经存有project_id: "HW-2024-001"和current_layer: "top_copper"；而 MCP Server 在收到请求时，并不需要自己去解析这些上下文，LangGraph 会在调用前自动将它们注入到 MCP 的params字段中。更重要的是，LangGraph 的conditional_edge（条件边）可以基于 MCP Server 返回的result.status字段，直接决定下一步是进入“仿真验证”节点，还是跳转到“人工审核”节点。这种基于结构化返回值的流程编排，是传统 HTTP 调用无法企及的。HTTP 只能告诉你“200 OK”或“500 Internal Error”，而 MCP 返回的{"status": "success", "data": {...}}或{"status": "error", "error_code": "insufficient_power_budget"}，才是 LangGraph 能读懂的“语言”。

2.3 “多 Server 调用”的本质：不是并发，而是“能力编排”，LangGraph 是唯一能驾驭它的框架

网络热词里频繁出现的“多 Server 调用”，常被误解为“同时调用多个服务”。这在 MCP 场景下是危险的。MCP 的设计初衷，是让 Agent 能像一个经验丰富的工程师一样，按需、有序、带上下文地调用不同的专业工具。比如，一个完整的硬件设计闭环可能是：先调用PCB Layout Server（生成初步布线）→ 根据其返回的estimated_power_consumption，判断是否需要优化 → 如果需要，则调用Power Analysis Server（进行精确功耗仿真）→ 最后，将两个 Server 的结果汇总，调用Report Generation Server（生成 PDF 报告）。这是一个清晰的、有依赖关系的 DAG（有向无环图），而不是一个并发的“大杂烩”。LangGraph 的graph.add_node()和graph.add_edge()正是为此而生。它强制你显式地定义每个 Server 的输入/输出契约，以及它们之间的数据流向。这种“声明式编排”，相比起在 FastAPI 的async def函数里手动await多个httpx.AsyncClient请求，其可维护性、可观测性和可测试性，提升了不止一个数量级。我亲眼见过一个项目，因为把所有外部调用都塞在一个async def里，导致一次Power Analysis Server的超时，直接拖垮了整个Report Generation流程，而用 LangGraph 后，我们只需要给Power Analysis节点设置一个timeout=30参数，超时后自动走fallback_edge到降级逻辑，主流程毫发无损。

3. 核心细节解析与实操要点：从协议握手到多 Server 调用的每一个“坑”

3.1 MCP 协议握手：不是简单的“ping-pong”，而是三次“能力契约确认”

MCP 的“握手”（Handshake）过程，远比想象中严谨。它不是客户端发个{"jsonrpc": "2.0", "method": "ping"}就完事了。真正的握手，是客户端和服务端之间，围绕一份机器可读、人可理解的能力契约（Capability Manifest）进行的三次交互。这三次交互，构成了整个 MCP 生态的信任基础。

第一次握手：客户端发起能力发现请求（list_capabilities）

客户端向 MCP Server 的/mcp端点发送一个标准的 JSON-RPC 2.0 请求：

{ "jsonrpc": "2.0", "id": 1, "method": "list_capabilities", "params": {} }

Server 必须返回一个包含所有可用能力的清单，每个能力都必须严格遵循 MCP 规范定义的CapabilitySchema。重点来了：这个返回体里，input_schema和output_schema字段，必须是完整的、可被 JSON Schema Validator 验证的 JSON Schema 对象。例如，一个用于查询芯片引脚信息的能力，其input_schema绝不能是模糊的{"type": "object", "properties": {"chip": {"type": "string"}}}，而必须是：

{ "type": "object", "properties": { "chip_model": { "type": "string", "enum": ["STM32F407VGT6", "ESP32-WROOM-32", "RP2040"] }, "pin_name": { "type": "string", "pattern": "^P[ABCD][0-15]$" } }, "required": ["chip_model", "pin_name"] }

注意：很多初学者在这里栽跟头。他们用pydantic.BaseModel.schema_json()生成的 Schema，往往缺少required字段，或者enum值是动态生成的，导致客户端无法在编译期就进行参数校验。正确的做法是，用pydantic.json_schema.model_json_schema()并传入mode="validation"参数，确保生成的 Schema 是为“校验”而非“序列化”服务的。

第二次握手：客户端发送能力注册请求（register_capability）

客户端拿到能力清单后，不会立刻调用。它会先向 Server 发送register_capability请求，表明自己“已知晓并接受该能力的契约”。这个请求的params字段，必须包含它所选择的capability_name和一个client_id。Server 收到后，会检查该能力是否允许被此client_id调用（实现权限控制），并返回一个registration_id。这个registration_id就像一张“临时工牌”，后续所有对该能力的调用，都必须携带它。这一步的设计，是为了防止恶意客户端随意探测和调用服务，是 MCP 安全模型的第一道防线。

第三次握手：客户端发起首次实际调用（call）

只有在成功完成前两次握手后，客户端才能发起真正的call请求。这个请求的结构是：

{ "jsonrpc": "2.0", "id": 3, "method": "call", "params": { "capability_name": "get_pin_info", "registration_id": "reg_abc123", "arguments": { "chip_model": "STM32F407VGT6", "pin_name": "PA0" } } }

Server 在收到后，会首先用registration_id查找对应的客户端权限，然后用input_schema严格校验arguments字段。任何校验失败，都会返回标准的invalid_parameter错误码，而不是一个模糊的400 Bad Request。这就是 MCP 所谓的“协议即契约”的体现——错误信息本身，就是协议的一部分，且是结构化的。

3.2 LangGraph 中的 MCP Server 封装：不是写一个函数，而是定义一个“状态感知的节点”

在 LangGraph 中封装一个 MCP Server，绝不是简单地def my_mcp_node(state): return mcp_client.call(...)。LangGraph 的强大之处，在于它要求你将“调用外部服务”这个动作，完全融入到整个状态机的生命周期中。这意味着，你需要定义一个@node装饰的函数，它接收State，并返回一个dict，这个dict的 key，必须和你定义的State类中的字段名完全一致。

假设我们有一个HardwareDesignState：

from typing import TypedDict, List, Optional class HardwareDesignState(TypedDict): project_id: str current_step: str pcb_layout_result: Optional[dict] power_analysis_result: Optional[dict] report_data: Optional[dict] error_log: List[str]

那么，封装一个调用PCB Layout Server的节点，应该是这样的：

from langgraph.graph import StateGraph from langgraph.prebuilt import ToolNode from mcp.client import MCPClient # 初始化 MCP 客户端（注意：这里用的是官方推荐的 async client） mcp_client = MCPClient("http://localhost:8000/mcp") @node async def run_pcb_layout(state: HardwareDesignState) -> dict: try: # 1. 从 state 中提取上下文，构造 MCP 调用参数 params = { "project_id": state["project_id"], "design_spec": { "board_size": "100x80mm", "layer_count": 4, "target_frequency": 100e6 } } # 2. 执行 MCP 调用（注意：这里是 await，因为 MCP client 是异步的） result = await mcp_client.call( capability_name="generate_pcb_layout", arguments=params, registration_id="reg_pcb_001" # 这个 ID 应该在初始化时就获取好 ) # 3. 将结构化结果，精准地映射回 state 的字段 return { "pcb_layout_result": result["data"], "current_step": "power_analysis", "error_log": [] # 清空之前的错误 } except MCPError as e: # 4. 将 MCP 的结构化错误，转化为 state 可理解的格式 return { "error_log": [f"MCP Error ({e.error_code}): {e.message}"], "current_step": "manual_review" }

实操心得：我踩过最大的一个坑，是在run_pcb_layout函数里，试图直接修改state字典，比如state["pcb_layout_result"] = result["data"]。这是完全错误的！LangGraph 的State是一个不可变的TypedDict，你只能通过return一个新字典来“更新”它。这个新字典里的 key，就是你要更新的 state 字段，value 就是新的值。LangGraph 会自动将这个字典“合并”到当前 state 中。这个设计看似麻烦，实则保证了状态流转的可预测性和可追溯性——每一次return，都是一次明确的、可审计的状态变更。

3.3 多 Server 调用的编排逻辑：用conditional_edge构建“智能决策树”

LangGraph 的add_conditional_edges方法，是实现“多 Server 调用”的灵魂。它允许你根据上一个节点的返回值，动态决定下一个节点。这正是 MCP 的结构化错误码和结果码大放异彩的地方。

继续上面的例子，run_pcb_layout节点执行完毕后，我们希望根据其返回的pcb_layout_result中的estimated_power字段，来决定下一步：

• 如果estimated_power < 5000（毫瓦），则直接进入generate_report节点。
• 如果5000 <= estimated_power < 10000，则先进入run_power_analysis节点。
• 如果estimated_power >= 10000，则跳转到manual_review节点。

这个逻辑，用 LangGraph 表达就是：

def decide_next_step(state: HardwareDesignState) -> str: """这是一个路由函数，它返回下一个节点的名字""" if state["error_log"]: return "manual_review" layout_result = state.get("pcb_layout_result") if not layout_result: return "manual_review" est_power = layout_result.get("estimated_power", 0) if est_power < 5000: return "generate_report" elif est_power < 10000: return "run_power_analysis" else: return "manual_review" # 构建图 graph = StateGraph(HardwareDesignState) # 添加节点 graph.add_node("run_pcb_layout", run_pcb_layout) graph.add_node("run_power_analysis", run_power_analysis) graph.add_node("generate_report", generate_report) graph.add_node("manual_review", manual_review) # 添加条件边：从 run_pcb_layout 节点出发，根据 decide_next_step 的返回值，走向不同节点 graph.add_conditional_edges( "run_pcb_layout", decide_next_step, { "run_power_analysis": "run_power_analysis", "generate_report": "generate_report", "manual_review": "manual_review" } ) # 添加普通边（无条件） graph.add_edge("run_power_analysis", "generate_report") graph.add_edge("generate_report", END) graph.add_edge("manual_review", END)

注意：decide_next_step函数的返回值，必须是字符串，且这个字符串必须是你图中已经add_node过的节点名。LangGraph 会严格校验这一点。这种“函数式路由”的设计，让你可以把复杂的业务决策逻辑，从节点内部剥离出来，放到一个独立的、可单元测试的函数里，极大地提升了代码的清晰度和可维护性。

4. 实操过程与核心环节实现：从零开始搭建 MCP + LangGraph 多 Server 系统

4.1 环境准备与工具链安装：避开 Python 版本和依赖冲突的深坑

我们将在 Ubuntu 22.04 (WSL2) 上进行部署。第一步，永远是环境隔离。绝对不要在系统 Python 或全局 pip 中安装 MCP 相关包。MCP 的生态目前还在快速迭代，不同版本的mcp、langgraph、langchain之间存在微妙的兼容性问题。

1. 创建专用虚拟环境

   # 创建一个名为 mcp-env 的虚拟环境 python3 -m venv ~/mcp-env # 激活它 source ~/mcp-env/bin/activate # 升级 pip 到最新版（避免旧版 pip 安装时出错） pip install --upgrade pip

2. 安装核心依赖（关键：指定版本）根据我实测，以下版本组合在生产环境中最为稳定：

   # 安装 LangGraph（注意：必须是 0.2.x，0.1.x 不支持最新的 MCP client） pip install langgraph==0.2.52 # 安装 MCP 官方客户端（这是最权威的实现） pip install mcp==0.1.12 # 安装 FastAPI 和 Uvicorn，用于构建第一个 Server pip install "fastapi[all]" uvicorn==0.29.0 # 安装 Pydantic v2，这是 MCP client 的硬性要求 pip install pydantic==2.7.1

   提示：mcp==0.1.12是一个关键版本。它修复了早期版本中register_capability请求在某些反向代理（如 Nginx）后面会丢失Content-Type头的问题。如果你用的是0.1.10或更早，可能会在握手阶段就卡在415 Unsupported Media Type错误上，查半天都找不到原因。

3. 验证安装

   python -c "import mcp; print(mcp.__version__)" python -c "import langgraph; print(langgraph.__version__)"

   输出应为0.1.12和0.2.52。如果报错，说明环境没激活或安装失败，务必重来。

4.2 构建第一个 MCP Server（FastAPI 版）：一个真实的 PCB 布线能力

我们将创建一个极简但功能完备的 MCP Server，它模拟一个 PCB 布线服务。它的核心是实现 MCP 规范要求的三个方法：list_capabilities、register_capability和call。

1. 创建项目目录结构

   mkdir -p ~/mcp-demo/server-pcb cd ~/mcp-demo/server-pcb touch main.py requirements.txt

2. 编写main.py

   from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import json from typing import Dict, Any, List, Optional app = FastAPI(title="PCB Layout MCP Server", version="1.0.0") # 存储已注册的 client_id 和其 registration_id 的映射（生产环境应换为 Redis） _registrations = {} # 定义能力契约（Manifest） CAPABILITIES = [ { "name": "generate_pcb_layout", "description": "Generates a preliminary PCB layout based on design specifications.", "input_schema": { "type": "object", "properties": { "project_id": {"type": "string"}, "design_spec": { "type": "object", "properties": { "board_size": {"type": "string"}, "layer_count": {"type": "integer", "minimum": 2, "maximum": 12}, "target_frequency": {"type": "number", "minimum": 1e6, "maximum": 10e9} }, "required": ["board_size", "layer_count", "target_frequency"] } }, "required": ["project_id", "design_spec"] }, "output_schema": { "type": "object", "properties": { "layout_id": {"type": "string"}, "estimated_power": {"type": "integer", "description": "Estimated power consumption in milliwatts"}, "routing_completion_rate": {"type": "number", "minimum": 0.0, "maximum": 1.0} }, "required": ["layout_id", "estimated_power", "routing_completion_rate"] } } ] @app.post("/mcp") async def mcp_endpoint(request: Request): try: body = await request.json() method = body.get("method") if method == "list_capabilities": return JSONResponse(content={ "jsonrpc": "2.0", "id": body.get("id"), "result": CAPABILITIES }) elif method == "register_capability": params = body.get("params", {}) client_id = params.get("client_id") capability_name = params.get("capability_name") if not client_id or not capability_name: raise HTTPException(400, "Missing client_id or capability_name") # 简单的注册逻辑：生成一个 registration_id reg_id = f"reg_{client_id}_{capability_name}_{hash(str(params)) % 10000}" _registrations[reg_id] = {"client_id": client_id, "capability_name": capability_name} return JSONResponse(content={ "jsonrpc": "2.0", "id": body.get("id"), "result": {"registration_id": reg_id} }) elif method == "call": params = body.get("params", {}) reg_id = params.get("registration_id") cap_name = params.get("capability_name") args = params.get("arguments", {}) if not reg_id or not cap_name or not args: raise HTTPException(400, "Missing registration_id, capability_name or arguments") # 验证 registration_id 是否有效 if reg_id not in _registrations: raise HTTPException(401, "Invalid registration_id") # 验证 capability_name 是否匹配 if _registrations[reg_id]["capability_name"] != cap_name: raise HTTPException(400, "Capability name mismatch") # 这里是核心业务逻辑：模拟 PCB 布线 if cap_name == "generate_pcb_layout": # 简单的模拟：根据 target_frequency 计算功耗 freq = args.get("design_spec", {}).get("target_frequency", 1e6) est_power = int(freq / 1e6 * 100) + 1000 # 毫瓦 result = { "layout_id": f"LAYOUT_{args['project_id']}_20240520", "estimated_power": est_power, "routing_completion_rate": 0.92 } return JSONResponse(content={ "jsonrpc": "2.0", "id": body.get("id"), "result": {"status": "success", "data": result} }) else: raise HTTPException(404, f"Unknown capability: {cap_name}") else: raise HTTPException(400, f"Unknown method: {method}") except json.JSONDecodeError: raise HTTPException(400, "Invalid JSON") except Exception as e: # MCP 要求所有错误都返回标准格式 return JSONResponse(content={ "jsonrpc": "2.0", "id": body.get("id") if 'body' in locals() else None, "error": { "code": -32603, # Internal error "message": str(e) } }, status_code=500) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

3. 启动 Server

   # 确保在 mcp-env 环境中 source ~/mcp-env/bin/activate cd ~/mcp-demo/server-pcb python main.py

   服务将在http://localhost:8000/mcp启动。你可以用curl测试第一次握手：

   curl -X POST http://localhost:8000/mcp \ -H "Content-Type: application/json" \ -d '{"jsonrpc": "2.0", "id": 1, "method": "list_capabilities", "params": {}}'

   你应该看到一个包含generate_pcb_layout能力的 JSON 响应。这标志着你的第一个 MCP Server 已经就绪。

4.3 构建 LangGraph 主流程：串联多个 Server 的“指挥中心”

现在，我们创建 LangGraph 的主程序，它将调用上面的server-pcb，并（为了演示）还调用一个假想的server-power（功率分析）。

1. 创建主程序目录

   mkdir -p ~/mcp-demo/agent-main cd ~/mcp-demo/agent-main touch main.py

2. 编写main.py

   from typing import TypedDict, List, Optional, Dict, Any from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver from mcp.client import MCPClient import asyncio # 1. 定义 State class HardwareDesignState(TypedDict): project_id: str current_step: str pcb_layout_result: Optional[Dict[str, Any]] power_analysis_result: Optional[Dict[str, Any]] report_data: Optional[Dict[str, Any]] error_log: List[str] # 2. 初始化 MCP Clients # 注意：这里我们为每个 Server 创建独立的 client pcb_client = MCPClient("http://localhost:8000/mcp") # power_client = MCPClient("http://localhost:8001/mcp") # 假设 power server 在 8001 # 3. 定义节点 @node async def run_pcb_layout(state: HardwareDesignState) -> dict: try: # 从 state 中提取参数 params = { "project_id": state["project_id"], "design_spec": { "board_size": "100x80mm", "layer_count": 4, "target_frequency": 100e6 } } # 执行 MCP 调用 # 注意：这里需要先完成 handshake！ # 我们在初始化 client 时，已经隐式完成了 list_capabilities 和 register_capability # 所以可以直接 call result = await pcb_client.call( capability_name="generate_pcb_layout", arguments=params ) # 返回更新后的 state return { "pcb_layout_result": result["data"], "current_step": "power_analysis", "error_log": [] } except Exception as e: return { "error_log": [f"PCB Layout Error: {str(e)}"], "current_step": "manual_review" } # 4. 定义路由函数 def decide_next_step(state: HardwareDesignState) -> str: if state["error_log"]: return "manual_review" layout_result = state.get("pcb_layout_result") if not layout_result: return "manual_review" est_power = layout_result.get("estimated_power", 0) if est_power < 5000: return "generate_report" elif est_power < 10000: return "run_power_analysis" else: return "manual_review" # 5. 构建图 graph = StateGraph(HardwareDesignState) # 添加节点 graph.add_node("run_pcb_layout", run_pcb_layout) # graph.add_node("run_power_analysis", run_power_analysis) # 留作扩展 # graph.add_node("generate_report", generate_report) # 留作扩展 graph.add_node("manual_review", lambda state: {"current_step": "done"}) # 添加条件边 graph.add_conditional_edges( "run_pcb_layout", decide_next_step, { # "run_power_analysis": "run_power_analysis", # "generate_report": "generate_report", "manual_review": "manual_review" } ) # 添加结束边 graph.add_edge("manual_review", END) # 设置入口点 graph.set_entry_point("run_pcb_layout") # 6. 编译图 app = graph.compile(checkpointer=MemorySaver()) # 7. 运行一个实例 if __name__ == "__main__": # 初始化初始状态 initial_state = { "project_id": "PROJ-001", "current_step": "start", "pcb_layout_result": None, "power_analysis_result": None, "report_data": None, "error_log": [] } # 运行 for output in app.stream(initial_state, stream_mode="values"): print("Current State:", output) print("Workflow completed.")

3. 运行主程序

   cd ~/mcp-demo/agent-main python main.py

   你会看到程序启动，向http://localhost:8000/mcp发起握手和调用，并最终打印出包含pcb_layout_result的状态。这证明 LangGraph 已经成功接管了 MCP Server 的调用。

4.4 多 Server 调用的终极形态：引入 Rust Server 与统一 MCP 网关

在真实世界中，你不可能让 LangGraph 的 Python 进程，直接去调用一个用 Rust 写的、运行在裸金属服务器上的硬件仿真器。这时，就需要一个MCP Gateway。它是一个轻量级的、语言无关的反向代理，它接收标准的 MCP 请求，根据capability_name，将其路由到后端不同的、物理隔离的 Server 上。

1. 架构图（文字描述）

   LangGraph (Python) | | Standard MCP JSON-RPC over HTTP v MCP Gateway (Rust, e.g., using `axum`) | |--- Route "simulate_hardware" --> Rust Hardware Simulator (on bare metal, port 8080) |--- Route "generate_pcb_layout" --> FastAPI Server (on localhost:8000) |--- Route "analyze_power" --> Node.js Power Analyzer (on docker:3000)

2. Gateway 的核心价值

   • 统一入口：LangGraph 只需要知道一个 URL (http://gateway:9000/mcp)，无需关心后端有多少个服务、它们用什么语言、部署在哪里。
   • 协议增强：Gateway 可以在转发前，自动添加认证头、日志记录、请求限流、甚至对arguments进行预处理（比如，将project_id映射为后端服务需要的tenant_id）。
   • 故障隔离：如果Rust Hardware Simulator崩溃了，Gateway 可以立即返回service_unavailable错误，而不会影响到FastAPI Server的调用。

3. 实操建议我们没有在本次 demo 中实现一个完整的 Gateway，但强烈建议你在项目初期就规划它。一个最小可行的 Gateway，可以用 Rust 的axum框架在 200 行代码内完成。它的核心逻辑就是一个match语句，根据params.capability_name，将请求reqwest::Client转发到对应的后端地址。这比在 LangGraph 的每个节点里硬编码一堆if-elif-else去判断capability_name并选择不同的MCPClient，要优雅和可维护得多。

5. 常见问题与排查技巧实录：那些只有亲手踩过才知道的“坑”

5.1 “Handshake failed: 415 Unsupported Media Type” —— Content-Type 的隐形杀手

现象：客户端在调用list_capabilities时，得到一个415错误，而不是预期的 JSON 响应。

根本原因：MCP 规范强制要求，所有请求的Content-Type必须是application/json。很多初学者在用curl测试时，会忘记加-H "Content-Type: application/json"。更隐蔽的情况是，当你用httpx.AsyncClient时，如果post方法没有显式指定headers，httpx默认不会发送Content-Type头，导致服务器拒绝