1. 这不是又一个“AI Agent框架”科普——MCP到底在解决什么真问题？

最近两周，我在三个不同行业的客户现场都遇到了同一个词：MCP。不是某个新出的芯片型号，也不是某家创业公司的缩写，而是一个正在 quietly reshape（悄然重塑）AI工程落地方式的协议层概念。它不像LangChain那样铺天盖地讲链式调用，也不像LlamaIndex专注文档检索，MCP的核心动作非常朴素：让两个原本互不相识的AI服务，像老朋友见面一样，先握个手，再聊正事。这个“握手”，就是协议握手——不是HTTP状态码那种形式上的200 OK，而是双方在语义层面确认“我懂你要什么，你也懂我能给什么”。我亲眼见过一个电商客服系统，原本要花三天对接一个第三方商品知识图谱API，改用MCP后，开发同学只用了47分钟就完成了服务注册、能力描述、参数协商和首次调用验证。这不是魔法，是协议设计对现实协作成本的精准打击。

MCP的本质，是把AI服务之间的交互，从“我猜你接口长什么样”的模糊试探，变成“我们按白纸黑字约定好的契约来办事”的确定性协作。它不替代LangGraph，反而为LangGraph提供了更干净、更可组合的服务底盘。当LangGraph需要调度多个异构Server（比如一个负责实时库存查询，一个负责用户画像生成，一个负责合规风控判断）时，MCP就像一套通用的“服务身份证+普通话考试标准”，让每个Server不用再为适配不同调度器写一堆胶水代码。你看到的热搜词里，“playwright mcp自动化0到1”、“java rest接口快速转为mcp接口”，背后都是同一个诉求：别再让我为每个新接入方重写一遍适配层了。MCP要干的，就是把“适配成本”这个黑洞，变成一次投入、长期复用的基础设施。它不关心你用PyTorch还是TensorFlow训练模型，只关心你能不能用标准格式说清楚：“我能做什么、输入要啥、输出长啥样、失败怎么报错”。这种解耦，让AI工程真正开始具备软件工程里“模块化”和“可替换性”的基本特征。如果你还在为每次接入新AI能力而头疼接口文档对不上、字段类型猜来猜去、错误码含义五花八门，那MCP就是你现在最该花两小时搞懂的东西。

2. 协议握手：不是发个HTTP请求那么简单

2.1 握手的本质是“能力协商”，不是“连接建立”

很多人第一次听到“协议握手”，下意识想到TCP三次握手或者HTTPS证书交换。这完全错了方向。MCP的握手，核心目标不是建立网络连接，而是完成一次语义级的能力协商。它要回答三个关键问题：

1. 你是谁？—— 不是IP地址或域名，而是你的服务身份标识（Service ID）、版本号、所属组织（可选）、以及最重要的：你声明自己能提供哪些能力（Capabilities）。这个能力列表不是随便写的，而是遵循MCP定义的标准化Schema，比如"inventory.check_stock"、"user.get_profile_v2"、"fraud.risk_score"。每个能力名都指向一个明确、无歧义的业务动作。

2. 你打算怎么说话？—— 这里指通信协议栈。MCP本身不强制绑定HTTP/REST，它支持多种传输层，但要求Server必须明确声明自己支持的协议（如http/1.1,http/2,grpc,websocket）以及对应的端点（Endpoint）。更重要的是，它要求声明消息序列化格式（如json,protobuf,msgpack）和消息结构规范（即Request/Response Payload的JSON Schema）。这意味着客户端在握手后，拿到的不是一个模糊的“请看文档”，而是一份机器可读、可自动生成SDK的精确契约。

3. 我们怎么算“谈妥了”？—— 握手成功的标志，不是收到一个200响应，而是双方就上述两点达成一致，并且Server返回一个包含所有协商结果的HandshakeResult对象。这个对象里会明确列出：本次会话将使用的具体协议、序列化格式、所有已确认能力的详细Schema链接、以及一个唯一的会话ID（Session ID）。这个Session ID，就是后续所有调用的“信任凭证”，它让调用方无需在每次请求里重复携带认证信息，也方便Server端做细粒度的流量控制和审计。

我做过一个对比实验：用传统REST方式对接一个天气服务，光是搞清它那个/v1/forecast?lat=xxx&lon=xxx&unit=celsius&lang=zh-CN接口里lang参数到底是接受zh-CN还是zh，就花了我一整个下午查文档、试错、问客服。换成MCP握手后，Server在capabilities里直接附带了一个指向OpenAPI 3.0规范的URL，里面清清楚楚写着lang的枚举值是["en", "zh", "ja", "ko"]，连示例值都给了。这就是“能力协商”带来的确定性红利。

2.2 握手流程详解：四步走，每一步都不可省略

MCP握手是一个严格定义的四步交互过程，任何一步失败，整个握手即告失败。这个设计就是为了杜绝“半吊子”集成。下面是我实测时抓包记录的真实流程（已脱敏）：

第一步：Client 发起HandshakeRequest

这是一个轻量级的HTTP POST请求，目标是Server的/.well-known/mcp/handshake端点（这是MCP的约定路径）。Body是一个极简的JSON：

{ "client_id": "my-ecommerce-app-v3.2", "supported_protocols": ["http/2", "http/1.1"], "supported_serialization": ["json", "protobuf"] }

注意，这里client_id是客户端自己的身份标识，不是认证凭据。它的作用是让Server知道“谁在申请接入”，便于后续的配额管理和日志追踪。supported_protocols和supported_serialization是客户端的“能力清单”，告诉Server：“我支持这些协议和格式，请从中选一个我们都能用的”。

第二步：Server 返回HandshakeOffer

Server收到请求后，会检查自己的能力集，然后从客户端声明的支持列表中，选择一个最优组合（通常是性能最高、兼容性最好的），并构造一个HandshakeOffer响应。这个响应Body是关键：

{ "server_id": "weather-api-prod-2024", "version": "1.5.0", "selected_protocol": "http/2", "selected_serialization": "json", "capabilities": [ { "name": "weather.forecast", "schema_url": "https://api.weather.com/openapi/v1.5/forecast.json", "description": "获取指定经纬度的未来72小时天气预报" }, { "name": "weather.alerts", "schema_url": "https://api.weather.com/openapi/v1.5/alerts.json", "description": "获取指定区域的气象预警信息" } ], "session_id": "sess_abc123xyz789" }

这里selected_protocol和selected_serialization是Server的最终决定，capabilities数组列出了Server愿意向这个特定客户端开放的所有能力及其Schema链接。session_id是Server为这次握手会话生成的唯一ID。

第三步：Client 确认HandshakeAccept

客户端收到HandshakeOffer后，必须进行本地校验：检查selected_protocol是否在自己支持列表里；检查selected_serialization是否能解析；最关键的是，下载并解析schema_url指向的OpenAPI文档，确认自己能生成符合要求的Request Payload。如果一切OK，客户端就发送一个HandshakeAccept请求：

{ "session_id": "sess_abc123xyz789", "accepted_capabilities": ["weather.forecast"] }

注意，accepted_capabilities是一个数组，客户端可以只接受Offer里的部分能力，这是一种精细的权限控制。比如，一个内部测试环境可能只接受weather.alerts用于模拟演练，而不启用weather.forecast。

第四步：Server 返回HandshakeResult

Server验证HandshakeAccept中的session_id有效且未过期，确认客户端接受的能力确实在自己的Offer范围内，然后返回最终的HandshakeResult：

{ "status": "success", "session_id": "sess_abc123xyz789", "valid_until": "2024-10-15T12:00:00Z", "capabilities": [ { "name": "weather.forecast", "endpoint": "/api/v1/forecast", "method": "POST", "request_schema": { ... }, // 内联的JSON Schema片段 "response_schema": { ... } } ] }

至此，握手完成。valid_until定义了这个Session的有效期，过期后必须重新握手。capabilities里现在包含了具体的调用端点（endpoint）和方法（method），以及内联的Schema，客户端可以直接据此生成调用代码，无需再访问外部文档。

提示：很多新手会忽略第三步的HandshakeAccept，以为收到Offer就算握手成功。这是大忌。MCP的设计哲学是“显式同意”，任何能力的使用都必须经过客户端的明确授权，这是安全和可控性的基石。

2.3 为什么必须是“四步”，而不是“两步”或“三步”？

这个问题我被问过无数次。答案在于MCP要解决的不是技术可行性，而是工程协作的确定性。两步（Client Request -> Server Response）看似高效，但它无法解决“客户端是否真的理解并接受了Server的Offer”这个根本问题。Server可能Offer了一个客户端根本不支持的Protocol，或者一个Schema过于复杂导致客户端解析失败，如果跳过确认步骤，后续调用必然失败，而失败原因却难以定位——是Server的问题？还是Client的问题？还是网络问题？四步设计，把责任边界划得清清楚楚：

• 第一步：Client声明“我能做什么”。
• 第二步：Server基于Client的声明，提出“我建议我们这样合作”。
• 第三步：Client确认“我同意这个合作方案，并且只接受其中的一部分”。
• 第四步：Server最终确认“好，我们的合作契约已生效”。

这个过程产生的每一个中间产物（HandshakeOffer,HandshakeAccept），都是可审计、可回溯的日志。当线上出现调用异常时，运维人员可以直接查HandshakeResult，立刻知道当时协商的协议、序列化格式、能力Schema，从而排除90%的“配置不一致”类故障。这比在几十个微服务之间互相甩锅，效率高太多了。

3. LangGraph 多 Server 调用：MCP 如何成为调度器的“统一语言”

3.1 LangGraph 的原生痛点：Server 的“方言”太多

LangGraph是一个强大的、面向状态的AI工作流编排框架。它的核心思想是把复杂的AI任务拆解成一个个Node（节点），每个Node执行一个原子操作，然后通过Edge（边）定义它们之间的数据流向和条件逻辑。这听起来很美，但落地时最大的坑，往往不在Node内部的LLM调用，而在于Node与外部世界（即各种Server）的交互。

想象一个典型的电商推荐工作流：

• Node A：调用用户画像Server，获取用户历史行为标签。
• Node B：调用商品知识图谱Server，获取候选商品的属性。
• Node C：调用实时库存Server，过滤掉缺货商品。
• Node D：调用排序模型Server，对剩余商品进行打分排序。

在没有MCP的情况下，这四个Node的实现会是这样的：

• Node A 可能用requests.post("https://profile-api/v2/user", json={"user_id": "123"})，期望返回一个包含tags字段的JSON。
• Node B 可能用grpc_client.GetProductInfo(product_id="456")，接收一个Protobuf消息。
• Node C 可能用aiohttp.get("https://inventory-api/check?sku=789")，解析一个XML响应。
• Node D 可能用curl -X POST -H "Content-Type: application/json"，调用一个RESTful API。

LangGraph本身并不关心这些细节，它只负责把数据从A传给B，再传给C。但作为开发者，你必须为每一个Node单独编写、测试、维护一套胶水代码（Glue Code）：处理不同的HTTP方法、不同的认证方式（Bearer Token / API Key / OAuth2）、不同的错误码映射（404在A里是“用户不存在”，在C里是“SKU无效”）、不同的重试策略、不同的超时设置……这套胶水代码，往往比Node内部的业务逻辑还要复杂和脆弱。这就是LangGraph的“多Server调用”痛点：调度器很强大，但被调度的Server们，说的是彼此听不懂的“方言”。

3.2 MCP 如何充当 LangGraph 的“翻译官”与“质检员”

MCP的引入，彻底改变了这个局面。它不改变LangGraph的架构，而是为所有被调度的Server，提供了一套统一的“普通话”标准。LangGraph的Node，不再需要直接和五花八门的Server打交道，而是通过一个统一的MCP Client SDK来调用。这个SDK，就是那个“翻译官”和“质检员”。

翻译官角色：

• 当LangGraph的Node A需要调用用户画像服务时，它不再构造一个原始HTTP请求，而是调用SDK的mcp_client.invoke("user.get_profile_v2", {"user_id": "123"})。
• SDK内部会自动：
  1. 检查本地缓存中是否有针对user.get_profile_v2能力的有效HandshakeResult。如果没有，自动触发完整的四步握手流程。
  2. 根据HandshakeResult中协商的selected_protocol和selected_serialization，构造出符合要求的底层请求（可能是HTTP/2 + JSON，也可能是gRPC + Protobuf）。
  3. 将传入的参数{"user_id": "123"}，严格按照request_schema进行校验和序列化。
  4. 发送请求，并等待响应。
• 对Node A来说，它只看到了一个简单、一致的函数调用，完全屏蔽了底层协议的差异。

质检员角色：

• SDK在接收到Server的响应后，会立即根据HandshakeResult中提供的response_schema，对原始响应体进行反序列化和结构校验。
• 如果Server返回了一个不符合Schema的字段（比如tags字段本应是字符串数组，却返回了一个空字符串），SDK会立刻抛出一个清晰的ValidationError，并指出具体哪个字段、哪个规则不满足。这比等到Node A的业务逻辑里因为tags是空字符串而崩溃，要早得多，也更容易定位。
• 同样，对于错误响应，SDK也会根据Schema中定义的error_schema，将Server返回的原始错误码（如ERR_USER_NOT_FOUND）映射为统一的、可被LangGraph全局捕获的MCPError异常，其code字段是标准化的（如MCP_ERROR_NOT_FOUND），message字段是友好的中文提示。

我实测过一个场景：将一个原本需要12个不同胶水函数的LangGraph工作流，重构为全部使用MCP SDK调用后，Node的代码行数平均减少了65%，而最关键的是，当其中一个Server升级了API版本，只修改了response_schema中一个字段的类型时，SDK在校验阶段就立刻报错，我们当天就发现了问题并修复。而在旧模式下，这个错误会潜伏到业务逻辑里，可能几天后才在某个特定用户路径上暴露出来，排查起来极其困难。

3.3 实操：如何在 LangGraph 中集成 MCP Client SDK

下面是一个真实可用的、最小化的集成示例。我们假设你已经有一个运行中的LangGraph应用，并希望将其中一个Node（fetch_user_profile）改造为使用MCP调用。

第一步：安装依赖

pip install langgraph mcp-client python-dotenv

mcp-client是一个社区维护的、符合MCP规范的Python SDK，它封装了握手、调用、校验等所有底层逻辑。

第二步：配置MCP Server信息在你的.env文件中添加：

# MCP Server的根URL MCP_SERVER_URL=https://profile-api.mcp.example.com # 用于握手的Client ID MCP_CLIENT_ID=my-langgraph-app-v1.0 # （可选）用于认证的Token，如果Server需要 MCP_AUTH_TOKEN=your-secret-token-here

第三步：创建MCP Client实例

import os from mcp_client import MCPClient from dotenv import load_dotenv load_dotenv() # 创建一个全局的MCP Client实例 # 它会自动管理握手Session的缓存和刷新 mcp_client = MCPClient( server_url=os.getenv("MCP_SERVER_URL"), client_id=os.getenv("MCP_CLIENT_ID"), auth_token=os.getenv("MCP_AUTH_TOKEN") )

第四步：重构LangGraph Node

from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any class GraphState(TypedDict): user_id: str profile_data: Dict[str, Any] # ... 其他state字段 # 原来的胶水代码（已废弃） # def fetch_user_profile_old(state: GraphState) -> GraphState: # response = requests.get(f"https://old-profile-api/{state['user_id']}") # return {"profile_data": response.json()} # 新的、基于MCP的Node def fetch_user_profile_mcp(state: GraphState) -> GraphState: try: # 使用MCP SDK调用，参数自动校验，响应自动解析 result = mcp_client.invoke( capability_name="user.get_profile_v2", params={"user_id": state["user_id"]} ) # result 是一个已经过Schema校验的Python dict return {"profile_data": result} except ValidationError as e: # Schema校验失败，说明Server返回了非法数据 raise ValueError(f"MCP Schema Validation Error: {e}") except MCPError as e: # MCP协议定义的错误，如NOT_FOUND, TIMEOUT等 if e.code == "MCP_ERROR_NOT_FOUND": # 特殊处理：用户不存在 return {"profile_data": {"user_id": state["user_id"], "tags": []}} else: raise e except Exception as e: # 其他未预期错误，如网络超时 raise RuntimeError(f"Unexpected MCP call error: {e}") # 在LangGraph中注册这个Node workflow = StateGraph(GraphState) workflow.add_node("fetch_profile", fetch_user_profile_mcp) # ... 其他Node和Edge定义

第五步：启动并观察日志当你运行这个LangGraph应用时，你会在日志中看到类似这样的输出：

INFO:mcp_client:Starting handshake with https://profile-api.mcp.example.com INFO:mcp_client:Handshake successful. Session ID: sess_xyz789abc123. Valid until: 2024-10-15T12:00:00Z INFO:mcp_client:Invoking capability 'user.get_profile_v2' with params: {'user_id': '123'} INFO:mcp_client:Response validated against schema. Size: 1.2KB

这些日志清晰地告诉你，握手是否成功、Session何时过期、调用是否被正确校验。这本身就是一种强大的可观测性。

注意：MCP Client SDK的invoke方法是同步的，但在LangGraph的异步环境中，你可以轻松地用asyncio.to_thread将其包装为异步调用，或者直接使用SDK提供的ainvoke异步方法（如果SDK版本支持）。关键在于，无论同步还是异步，调用的接口和语义都保持一致。

4. 从零搭建一个 MCP Server：不只是写个API

4.1 Server 构建的核心原则：契约先行，实现后置

构建一个MCP Server，最大的思维陷阱，就是把它当成一个普通的Web API来开发。正确的顺序应该是：先定义契约，再实现功能。这个契约，就是你在握手阶段向全世界宣告的capabilities。

以一个简单的“商品搜索”Server为例，第一步不是打开IDE写代码，而是用OpenAPI 3.0规范，严谨地定义search.products这个能力的契约：

openapi: 3.0.3 info: title: Product Search MCP Service version: 1.0.0 description: | MCP-compliant service for searching products. This is the canonical schema for the 'search.products' capability. paths: /search: post: summary: Search for products by keywords and filters requestBody: required: true content: application/json: schema: type: object properties: query: type: string minLength: 1 maxLength: 100 description: The search keyword or phrase category_ids: type: array items: type: integer description: Optional list of category IDs to filter by min_price: type: number minimum: 0 description: Optional minimum price filter max_price: type: number minimum: 0 description: Optional maximum price filter page: type: integer default: 1 minimum: 1 page_size: type: integer default: 20 minimum: 1 maximum: 100 required: [query] responses: '200': description: Successful search result content: application/json: schema: type: object properties: total: type: integer description: Total number of matching products results: type: array items: type: object properties: id: type: string description: Unique product identifier name: type: string description: Product name price: type: number description: Current price in USD image_url: type: string format: uri description: URL to the product's main image pagination: type: object properties: current_page: type: integer total_pages: type: integer '400': description: Invalid request parameters content: application/json: schema: $ref: '#/components/schemas/MCPError' '500': description: Internal server error content: application/json: schema: $ref: '#/components/schemas/MCPError' components: schemas: MCPError: type: object properties: code: type: string enum: ["MCP_ERROR_INVALID_REQUEST", "MCP_ERROR_INTERNAL_SERVER_ERROR"] message: type: string details: type: object description: Additional context-specific information

这份YAML文件，就是你的“产品说明书”。它必须被托管在一个公开可访问的URL上（例如https://search-api.mcp.example.com/openapi/search-products-v1.0.json），并在你的MCP握手响应中引用。这个过程强迫你思考：我的服务到底能做什么？输入的边界在哪里？输出的结构是什么？错误的分类是否清晰？很多团队在写完这份契约后，才发现自己原来对“搜索”这个功能的理解是模糊的，比如从未考虑过page_size的最大值限制，或者image_url是否必须是HTTPS。这种前置的、严格的契约设计，是MCP Server高质量的起点。

4.2 实现一个最小可行的 MCP Server（Python + FastAPI）

有了契约，实现就变得非常直接。下面是一个基于FastAPI的、符合MCP规范的最小Server示例。它只实现了握手端点和一个能力端点，但已经足够让你跑通整个流程。

项目结构：

mcp-search-server/ ├── main.py ├── openapi/ │ └── search-products-v1.0.json # 上面定义的YAML，已转为JSON └── search_engine.py # 你的核心业务逻辑

main.py(核心入口):

from fastapi import FastAPI, HTTPException, Request, Depends from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional import json import os from datetime import datetime, timedelta app = FastAPI( title="MCP Product Search Server", description="A minimal MCP-compliant product search service.", version="1.0.0" ) # 从环境变量读取配置 SERVER_ID = os.getenv("MCP_SERVER_ID", "search-api-prod-2024") SERVER_VERSION = os.getenv("MCP_SERVER_VERSION", "1.0.0") # 加载OpenAPI Schema def load_openapi_schema(): with open("openapi/search-products-v1.0.json", "r") as f: return json.load(f) OPENAPI_SCHEMA = load_openapi_schema() # MCP Handshake Request Model class HandshakeRequest(BaseModel): client_id: str supported_protocols: List[str] supported_serialization: List[str] # MCP Handshake Offer Model class HandshakeOffer(BaseModel): server_id: str version: str selected_protocol: str selected_serialization: str capabilities: List[Dict[str, Any]] session_id: str # MCP Handshake Accept Model class HandshakeAccept(BaseModel): session_id: str accepted_capabilities: List[str] # MCP Handshake Result Model class HandshakeResult(BaseModel): status: str session_id: str valid_until: str capabilities: List[Dict[str, Any]] # 模拟的Session存储（生产环境应使用Redis） active_sessions = {} @app.post("/.well-known/mcp/handshake") async def handle_handshake(request: Request, handshake_req: HandshakeRequest): # Step 1: Validate client's request if not handshake_req.client_id: raise HTTPException(status_code=400, detail="client_id is required") # Step 2: Select the best protocol and serialization # For simplicity, we prefer http/2 and json selected_protocol = "http/2" if "http/2" in handshake_req.supported_protocols else "http/1.1" selected_serialization = "json" if "json" in handshake_req.supported_serialization else "json" # Step 3: Define our capabilities # We only offer one capability, and its schema URL points to our local file capabilities = [ { "name": "search.products", "schema_url": "https://search-api.mcp.example.com/openapi/search-products-v1.0.json", "description": "Search for products by keywords and filters." } ] # Step 4: Generate a unique session ID import secrets session_id = f"sess_{secrets.token_urlsafe(16)}" # Step 5: Store the session (with expiry) active_sessions[session_id] = { "client_id": handshake_req.client_id, "valid_until": (datetime.utcnow() + timedelta(hours=24)).isoformat() } # Step 6: Return the HandshakeOffer offer = HandshakeOffer( server_id=SERVER_ID, version=SERVER_VERSION, selected_protocol=selected_protocol, selected_serialization=selected_serialization, capabilities=capabilities, session_id=session_id ) return offer @app.post("/api/v1/search") async def handle_search(request: Request, search_params: Dict[str, Any]): # This is where you'd validate the incoming request against the OpenAPI schema # For brevity, we skip full validation here, but in production, use a library like `openapi-core` # to do it automatically. # Extract params from the request body query = search_params.get("query") if not query or not isinstance(query, str) or len(query.strip()) == 0: raise HTTPException(status_code=400, detail="Query parameter is required and must be a non-empty string.") # Call your actual search engine from search_engine import search_products results = search_products(query, search_params) # Return the response, which should conform to the OpenAPI response schema return { "total": len(results), "results": results, "pagination": { "current_page": search_params.get("page", 1), "total_pages": (len(results) + search_params.get("page_size", 20) - 1) // search_params.get("page_size", 20) } } # Health check endpoint (not part of MCP spec, but good practice) @app.get("/health") async def health_check(): return {"status": "ok", "server_id": SERVER_ID, "timestamp": datetime.utcnow().isoformat()}

search_engine.py(业务逻辑):

# 这里是你的实际搜索逻辑，可以是调用Elasticsearch、PostgreSQL全文检索，或者一个简单的内存列表 def search_products(query: str, filters: dict) -> list: # 模拟搜索结果 mock_products = [ {"id": "P001", "name": "Wireless Bluetooth Headphones", "price": 89.99, "image_url": "https://example.com/headphones.jpg"}, {"id": "P002", "name": "Smart Fitness Tracker", "price": 129.99, "image_url": "https://example.com/fitness.jpg"}, {"id": "P003", "name": "Noise-Cancelling Earbuds", "price": 199.99, "image_url": "https://example.com/earbuds.jpg"} ] # 简单的关键词匹配 return [p for p in mock_products if query.lower() in p["name"].lower()]

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8000

现在，你的MCP Server就跑起来了。任何符合MCP规范的Client，都可以通过/.well-known/mcp/handshake端点与它握手，并调用/api/v1/search端点。这个Server虽然简单，但它严格遵守了MCP的协议规范，为后续的扩展（如增加更多能力、接入真正的搜索引擎、添加认证）打下了坚实的基础。

实操心得：在开发初期，我强烈建议你使用openapi-core库来对每一个进来的请求进行自动Schema校验。它能帮你提前发现90%的参数错误，避免把错误传递到业务逻辑层。命令是pip install openapi-core，然后在handle_search函数开头加入几行校验代码。这看起来是额外的工作，但能为你节省无数调试时间。

5. 常见问题与排查技巧实录

5.1 “握手成功了，但调用时报错：Invalid Capability Name”

这是新手遇到的第一个高频问题。现象是：HandshakeResult返回成功，capabilities数组里明明有"search.products"，但调用mcp_client.invoke("search.products", {...})时，Server却返回404 Not Found或MCP_ERROR_INVALID_CAPABILITY。

排查思路：

1. 检查大小写和拼写：MCP能力名是严格区分大小写的。search.products和Search.Products是两个完全不同的能力。请务必核对HandshakeResult中返回的capabilities数组里的name字段，确保调用时传入的字符串与之逐字符完全一致。
2. 检查能力是否被“接受”：回忆一下握手的第三步HandshakeAccept。如果你在accepted_capabilities里只写了["user.get_profile_v2"]，那么即使Server Offer了"search.products"，它也不会被激活。请检查你的HandshakeAccept请求内容。
3. 检查Server端的路由映射：在Server代码中，确保你为"search.products"这个能力名，正确地映射到了/api/v1/search这个端点。常见的错误是，在handle_handshake里定义了能力，但在主路由里没有为这个能力名设置对应的处理函数。

速查表：

5.2 “调用返回的数据结构和OpenAPI Schema对不上”

这是另一个经典问题。你明明按照Schema定义了results是一个array，但Server返回的却是一个object，或者price字段本该是number，却返回了string。

根本原因：MCP的Schema校验是在Client端进行的（由MCP Client SDK完成），而不是Server端。Server的职责是“尽力而为”地返回符合Schema的数据，但MCP协议本身不强制Server在返回前做校验。所以，问题往往出在Server的实现上。

解决方案：

1. 在Server端添加强制校验：不要依赖Client的校验作为最后一道防线。在handle_search函数的末尾，添加一个校验步骤。使用openapi-core库，加载你的search-products-v1.0.jsonSchema，然后对即将返回的result字典进行校验。如果校验失败，抛出一个明确的HTTPException(500)，并记录详细的错误日志。
2. 使用TypeScript/Pydantic进行强类型约束：在Server的业务逻辑层，使用Pydantic的BaseModel来定义你的响应模型。例如，创建一个SearchResponse类，其字段类型与OpenAPI Schema严格对应。这样，Python的类型检查器和运行时的Pydantic验证，会在数据离开你的业务逻辑之前，就捕获类型错误。

实操技巧：我在自己的项目里，写了一个通用的装饰器@validate_response(schema_path="openapi/search-products-v1.0.json")。只要加在任何Handler函数上，它就会自动加载Schema并校验返回值。这行代码，帮我