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的握手核心目标不是建立网络连接而是完成一次语义级的能力协商。它要回答三个关键问题你是谁—— 不是IP地址或域名而是你的服务身份标识Service ID、版本号、所属组织可选、以及最重要的你声明自己能提供哪些能力Capabilities。这个能力列表不是随便写的而是遵循MCP定义的标准化Schema比如inventory.check_stock、user.get_profile_v2、fraud.risk_score。每个能力名都指向一个明确、无歧义的业务动作。你打算怎么说话—— 这里指通信协议栈。MCP本身不强制绑定HTTP/REST它支持多种传输层但要求Server必须明确声明自己支持的协议如http/1.1,http/2,grpc,websocket以及对应的端点Endpoint。更重要的是它要求声明消息序列化格式如json,protobuf,msgpack和消息结构规范即Request/Response Payload的JSON Schema。这意味着客户端在握手后拿到的不是一个模糊的“请看文档”而是一份机器可读、可自动生成SDK的精确契约。我们怎么算“谈妥了”—— 握手成功的标志不是收到一个200响应而是双方就上述两点达成一致并且Server返回一个包含所有协商结果的HandshakeResult对象。这个对象里会明确列出本次会话将使用的具体协议、序列化格式、所有已确认能力的详细Schema链接、以及一个唯一的会话IDSession ID。这个Session ID就是后续所有调用的“信任凭证”它让调用方无需在每次请求里重复携带认证信息也方便Server端做细粒度的流量控制和审计。我做过一个对比实验用传统REST方式对接一个天气服务光是搞清它那个/v1/forecast?latxxxlonxxxunitcelsiuslangzh-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 返回HandshakeOfferServer收到请求后会检查自己的能力集然后从客户端声明的支持列表中选择一个最优组合通常是性能最高、兼容性最好的并构造一个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 返回HandshakeResultServer验证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_id456)接收一个Protobuf消息。Node C 可能用aiohttp.get(https://inventory-api/check?sku789)解析一个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内部会自动检查本地缓存中是否有针对user.get_profile_v2能力的有效HandshakeResult。如果没有自动触发完整的四步握手流程。根据HandshakeResult中协商的selected_protocol和selected_serialization构造出符合要求的底层请求可能是HTTP/2 JSON也可能是gRPC Protobuf。将传入的参数{user_id: 123}严格按照request_schema进行校验和序列化。发送请求并等待响应。对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_FOUNDmessage字段是友好的中文提示。我实测过一个场景将一个原本需要12个不同胶水函数的LangGraph工作流重构为全部使用MCP SDK调用后Node的代码行数平均减少了65%而最关键的是当其中一个Server升级了API版本只修改了response_schema中一个字段的类型时SDK在校验阶段就立刻报错我们当天就发现了问题并修复。而在旧模式下这个错误会潜伏到业务逻辑里可能几天后才在某个特定用户路径上暴露出来排查起来极其困难。3.3 实操如何在 LangGraph 中集成 MCP Client SDK下面是一个真实可用的、最小化的集成示例。我们假设你已经有一个运行中的LangGraph应用并希望将其中一个Nodefetch_user_profile改造为使用MCP调用。第一步安装依赖pip install langgraph mcp-client python-dotenvmcp-client是一个社区维护的、符合MCP规范的Python SDK它封装了握手、调用、校验等所有底层逻辑。第二步配置MCP Server信息在你的.env文件中添加# MCP Server的根URL MCP_SERVER_URLhttps://profile-api.mcp.example.com # 用于握手的Client ID MCP_CLIENT_IDmy-langgraph-app-v1.0 # 可选用于认证的Token如果Server需要 MCP_AUTH_TOKENyour-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_urlos.getenv(MCP_SERVER_URL), client_idos.getenv(MCP_CLIENT_ID), auth_tokenos.getenv(MCP_AUTH_TOKEN) )第四步重构LangGraph Nodefrom 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(fhttps://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_nameuser.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(fMCP 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(fUnexpected 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不只是写个API4.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 products 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 ServerPython 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( titleMCP Product Search Server, descriptionA minimal MCP-compliant product search service., version1.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 clients request if not handshake_req.client_id: raise HTTPException(status_code400, detailclient_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 fsess_{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(hours24)).isoformat() } # Step 6: Return the HandshakeOffer offer HandshakeOffer( server_idSERVER_ID, versionSERVER_VERSION, selected_protocolselected_protocol, selected_serializationselected_serialization, capabilitiescapabilities, session_idsession_id ) return offer app.post(/api/v1/search) async def handle_search(request: Request, search_params: Dict[str, Any]): # This is where youd 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_code400, detailQuery 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。排查思路检查大小写和拼写MCP能力名是严格区分大小写的。search.products和Search.Products是两个完全不同的能力。请务必核对HandshakeResult中返回的capabilities数组里的name字段确保调用时传入的字符串与之逐字符完全一致。检查能力是否被“接受”回忆一下握手的第三步HandshakeAccept。如果你在accepted_capabilities里只写了[user.get_profile_v2]那么即使Server Offer了search.products它也不会被激活。请检查你的HandshakeAccept请求内容。检查Server端的路由映射在Server代码中确保你为search.products这个能力名正确地映射到了/api/v1/search这个端点。常见的错误是在handle_handshake里定义了能力但在主路由里没有为这个能力名设置对应的处理函数。速查表现象最可能原因快速验证方法HandshakeResult里有search.products但调用时报404Server端没有为该能力名注册路由直接用curl访问/api/v1/search看是否返回正常响应HandshakeResult里没有search.productsServer的HandshakeOffer中遗漏了该能力查看Server日志确认handle_handshake函数返回的capabilities数组内容HandshakeResult里有但调用时报MCP_ERROR_INVALID_CAPABILITYHandshakeAccept中未包含该能力名检查HandshakeAccept请求的accepted_capabilities字段5.2 “调用返回的数据结构和OpenAPI Schema对不上”这是另一个经典问题。你明明按照Schema定义了results是一个array但Server返回的却是一个object或者price字段本该是number却返回了string。根本原因MCP的Schema校验是在Client端进行的由MCP Client SDK完成而不是Server端。Server的职责是“尽力而为”地返回符合Schema的数据但MCP协议本身不强制Server在返回前做校验。所以问题往往出在Server的实现上。解决方案在Server端添加强制校验不要依赖Client的校验作为最后一道防线。在handle_search函数的末尾添加一个校验步骤。使用openapi-core库加载你的search-products-v1.0.jsonSchema然后对即将返回的result字典进行校验。如果校验失败抛出一个明确的HTTPException(500)并记录详细的错误日志。使用TypeScript/Pydantic进行强类型约束在Server的业务逻辑层使用Pydantic的BaseModel来定义你的响应模型。例如创建一个SearchResponse类其字段类型与OpenAPI Schema严格对应。这样Python的类型检查器和运行时的Pydantic验证会在数据离开你的业务逻辑之前就捕获类型错误。实操技巧我在自己的项目里写了一个通用的装饰器validate_response(schema_pathopenapi/search-products-v1.0.json)。只要加在任何Handler函数上它就会自动加载Schema并校验返回值。这行代码帮我
