摘要AI对话平台集成MCP Server的桥接方案详解通过MCP-Proxy和本地桥接服务让AI对话平台调用自定义MCP工具扩展AI对话平台的工具调用能力。第41篇 AI助手集成 通过桥接让AI对话平台使用MCP标签: MCP, AI助手, 桥接, MCP集成, AI平台前几天有个朋友问我他公司用 MCP 写了一堆内部工具服务器有查订单的、查库存的、查日志的现在想把这些工具接到 AI 对话平台上用问有没有什么办法。我说这个我有经验去年我就干过这事儿中间踩了个特别坑的坑差点把项目搞黄了。为什么要用桥接先说背景。某主流AI对话平台虽然在 2025 年 3 月开始支持了原生的 MCP 连接器Connectors但这个功能有限制。它只支持远程的 Streamable HTTP 传输方式而且得是平台方审核过的连接器。你自己写的本地 MCP Server 用的是 stdio 传输AI对话平台根本连不上。那怎么办呢思路其实很简单。该AI平台有一个自定义助手功能你可以创建自定义助手给它配置Actions。Actions本质上就是让助手调用外部 HTTP API。我们只要写一个桥接服务把 MCP Server 的工具转换成 HTTP API然后把 API 的 OpenAPI Schema 导入到自定义助手里AI对话平台就能间接地调用 MCP 工具了。整个架构是这样的AI对话平台发 HTTP 请求到桥接服务桥接服务通过 stdio 或者 SSE 跟 MCP Server 通信拿到结果后转成 HTTP 响应返回给AI对话平台。桥接架构设计我把这个桥接架构分成了三层。第一层是 AI平台侧自定义助手通过Actions发送 HTTP 请求。每个 MCP 工具对应一个 HTTP 端点。第二层是桥接服务用 Python 写的 FastAPI 应用。它做三件事一是启动并管理 MCP Server 子进程二是把 MCP 的 JSON-RPC 协议转换成 HTTP 的 REST 接口三是自动生成 OpenAPI Schema 供自定义助手导入。第三层是 MCP Server 本身可以是任何用 MCP SDK 写的 Serverstdio 传输方式即可。我画了个简单的流程图帮你们理解。用户在AI对话平台中提问 | v AI对话平台分析后决定调用某个Action | v AI对话平台向桥接服务发送 HTTP POST 请求 | v 桥接服务收到请求后通过 stdio 向 MCP Server 发 JSON-RPC 消息 | (tools/call) v MCP Server 执行工具返回结果 | v 桥接服务把结果转成 HTTP JSON 响应 | v AI对话平台拿到结果组织自然语言回复这个架构的好处是MCP Server 不需要做任何改动桥接服务帮你完成了协议转换。你已有的 MCP Server 可以直接用上。构建桥接服务下面是完整的桥接服务代码用 Python 写的基于 FastAPI 和 MCP Python SDK。# bridge_server.py - MCP 到AI对话平台的桥接服务# 依赖: pip install fastapi uvicorn mcp httpximportasyncioimportjsonimportosfromcontextlibimportasynccontextmanagerfromtypingimportAnyfromfastapiimportFastAPI,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefrompydanticimportBaseModelfrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_client# 全局配置 # MCP Server 的启动配置可以根据需要修改MCP_SERVER_CONFIG{command:python,# 启动命令args:[mcp_server.py],# 启动参数指向你的 MCP Server 文件env:{# 环境变量API_KEY:os.getenv(API_KEY,),},}# 全局的 MCP Client Session 引用在 lifespan 中初始化mcp_session:ClientSession|NoneNone# 缓存工具列表避免每次请求都去查cached_tools:list[dict][]# 生命周期管理 asynccontextmanagerasyncdeflifespan(app:FastAPI): FastAPI 的生命周期管理器 应用启动时创建 MCP 连接应用关闭时断开连接 globalmcp_session,cached_tools# 构建 stdio 连接参数server_paramsStdioServerParameters(commandMCP_SERVER_CONFIG[command],argsMCP_SERVER_CONFIG[args],envMCP_SERVER_CONFIG[env],)# 启动 MCP Server 子进程并建立连接asyncwithstdio_client(server_params)as(read,write):# 创建 Client SessionasyncwithClientSession(read,write)assession:# 执行 MCP 握手initializeawaitsession.initialize()# 获取工具列表并缓存tools_resultawaitsession.list_tools()cached_tools[{name:tool.name,description:tool.description,inputSchema:tool.inputSchema,}fortoolintools_result.tools]# 把 session 存到全局变量供路由使用mcp_sessionsession# 应用运行期间保持连接yield# 应用关闭时清理mcp_sessionNone# FastAPI 应用 # 创建 FastAPI 应用指定生命周期管理器appFastAPI(titleMCP Bridge,description将 MCP Server 的工具桥接为 HTTP API 供AI对话平台调用,version1.0.0,lifespanlifespan,)# 添加 CORS 支持AI对话平台的请求需要跨域app.add_middleware(CORSMiddleware,allow_origins[*],# 生产环境应限制为你的平台域名allow_credentialsTrue,allow_methods[*],allow_headers[*],)# API 路由 classToolCallRequest(BaseModel): 工具调用请求的 body 模型 AI对话平台调用Action时会发送这个格式的请求 arguments:dict[str,Any]{}# 工具参数键值对形式app.get(/openapi.json)asyncdefget_openapi_schema(): 返回 OpenAPI 3.0 Schema 自定义助手在配置Action时需要导入这个schema 我们把每个 MCP 工具动态生成为一个 POST 端点 # 获取 FastAPI 默认生成的 OpenAPI schemaschemaapp.openapi()# 遍历缓存的工具列表手动添加每个工具对应的端点描述fortoolincached_tools:tool_nametool[name]pathf/tools/{tool_name}# 构建每个工具的 OpenAPI 路径定义schema[paths][path]{post:{summary:tool[description]ortool_name,description:fMCP 工具:{tool_name},operationId:fcall_{tool_name},requestBody:{required:False,content:{application/json:{schema:{type:object,properties:tool[inputSchema].get(properties,{}),}}},},responses:{200:{description:工具调用结果,content:{application/json:{schema:{type:string}}},}},}}returnschemaapp.get(/tools)asyncdeflist_tools(): 列出所有可用的 MCP 工具 方便调试时查看当前桥接了哪些工具 return{tools:cached_tools}app.post(/tools/{tool_name})asyncdefcall_tool(tool_name:str,request:ToolCallRequest): 调用指定的 MCP 工具 这是自定义助手Actions实际会调用的端点 ifmcp_sessionisNone:# Session 未初始化说明 MCP Server 连接出了问题raiseHTTPException(status_code503,detailMCP Server 未连接)# 检查工具是否存在tool_names[t[name]fortincached_tools]iftool_namenotintool_names:raiseHTTPException(status_code404,detailf工具{tool_name}不存在可用工具:{tool_names})try:# 调用 MCP 工具传入参数resultawaitmcp_session.call_tool(tool_name,request.arguments)# 提取文本内容返回texts[]forcontentinresult.content:ifhasattr(content,text):texts.append(content.text)return{result:\n.join(texts)}exceptExceptionase:# 工具调用出错时返回 500raiseHTTPException(status_code500,detailf工具调用失败:{str(e)})app.get(/health)asyncdefhealth_check(): 健康检查端点 用于验证桥接服务是否正常运行 return{status:ok,mcp_connected:mcp_sessionisnotNone,tools_count:len(cached_tools),}# 启动入口 if__name____main__:importuvicorn# 启动 FastAPI 服务监听 8000 端口# 生产环境建议用 0.0.0.0 并配置 HTTPSuvicorn.run(app,host0.0.0.0,port8000,)这段代码的核心逻辑是启动时通过 MCP SDK 的stdio_client连接 MCP Server完成握手后缓存工具列表。然后为每个工具动态注册一个 HTTP POST 端点。同时提供一个/openapi.json端点返回包含所有工具定义的 OpenAPI Schema。配合使用的 MCP Server为了让桥接服务有东西可连我写了一个简单的 MCP Server 作为示例。这个 Server 暴露两个工具一个查天气一个查股票。# mcp_server.py - 示例 MCP Server提供天气和股票查询工具# 依赖: pip install mcp httpximportasyncioimporthttpxfrommcp.serverimportServerfrommcp.server.stdioimportstdio_serverfrommcp.typesimport(Tool,TextContent,)# 创建 MCP Server 实例serverServer(weather-stock-server)server.list_tools()asyncdeflist_tools()-list[Tool]: 返回此 Server 支持的所有工具 每个工具包含名称、描述和输入参数 schema return[Tool(nameget_weather,description查询指定城市的天气信息,inputSchema{type:object,properties:{city:{type:string,description:城市名称例如 Beijing,},},required:[city],},),Tool(nameget_stock_price,description查询指定股票代码的当前价格,inputSchema{type:object,properties:{symbol:{type:string,description:股票代码例如 AAPL,},},required:[symbol],},),]server.call_tool()asyncdefcall_tool(name:str,arguments:dict)-list[TextContent]: 工具调用处理器 根据工具名分发到对应的处理函数 ifnameget_weather:# 从参数中提取城市名cityarguments.get(city,unknown)# 这里用模拟数据实际项目可以调真实天气 APIweather_data{Beijing:晴天 25度,Shanghai:多云 28度,Guangzhou:雷阵雨 30度,}resultweather_data.get(city,f暂无{city}的天气数据)return[TextContent(typetext,textf{city}天气:{result})]elifnameget_stock_price:# 从参数中提取股票代码symbolarguments.get(symbol,unknown)# 模拟查询股票价格stock_prices{AAPL:$185.32,GOOGL:$142.65,TSLA:$248.50,}resultstock_prices.get(symbol,f未找到{symbol}的价格)return[TextContent(typetext,textf{symbol}当前价格:{result})]# 未知工具返回错误提示return[TextContent(typetext,textf未知工具:{name})]asyncdefmain(): 主函数启动 stdio Server asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream)if__name____main__:asyncio.run(main())OpenAPI Schema 生成与导入桥接服务跑起来之后访问http://localhost:8000/openapi.json就能拿到自动生成的 OpenAPI Schema。但这里有个问题自定义助手要求导入的 OpenAPI Schema 必须是公网可访问的 URL。你本地跑的服务AI对话平台是访问不到的。所以你需要把桥接服务部署到公网服务器上。可以用云服务器、Vercel、Railway 之类的平台。部署好之后拿到公网 URL比如https://my-bridge.example.com。然后在AI对话平台里创建一个新的自定义助手步骤如下。第一步进入AI对话平台点左上角的探索选择创建助手。第二步在 配置页面里往下翻找到Actions设置点创建新Action。第三步在 OpenAPI Schema那里选从URL导入填入https://my-bridge.example.com/openapi.json。第四步AI对话平台会自动解析你的Schema把每个工具显示为一个Action。你可以逐个测试这些 Action 是否能正常调用。第五步如果一切正常保存 自定义助手然后在对话里就可以使用这些 MCP 工具了。比如你可以问北京今天天气怎么样AI对话平台会自动调用get_weather这个Action。与原生 MCP 客户端的对比桥接方案虽然能用但跟原生 MCP 客户端比如 Claude Desktop比起来差异还是挺大的。对比维度桥接方案 (自定义助手)原生 MCP (Claude Desktop)传输方式HTTP REST APIstdio / SSE协议转换需要 MCP 到 HTTP 的转换层直接使用 JSON-RPC部署复杂度需要公网服务器 HTTPS本地运行即可延迟较高多了 HTTP 往返低本地进程通信工具发现静态需手动导入 Schema动态实时 tools/list工具更新需重新导入 Schema 才能生效自动发现新工具认证通过 API Key 或 OAuth无需额外认证会话管理无状态每次请求独立有状态可保持上下文调试便利性需要查服务端日志客户端直接显示调用过程成本需要服务器费用无额外成本从表格能看出来桥接方案最大的问题是延迟和工具发现的静态性。每次 MCP Server 加了新工具你都得重新导入 Schema。而且因为走的是公网 HTTP延迟比本地通信高不少。独家踩坑经验 OpenAPI Schema 格式不兼容导致 Action 调用失败这个坑差点让我项目延期分享出来希望你们别再踩。事情是这样的我的桥接服务跑起来后/openapi.json端点返回的 Schema 在本地用 Swagger UI 看完全正常所有端点都能显示。但是导入到自定义助手的时候报了一个错说 Schema 格式有问题。我一开始以为是 OpenAPI 版本的问题FastAPI 默认生成的是 3.1.0 版本而 AI对话平台可能只支持3.0.0。我改了一下 FastAPI 的配置强制输出 3.0 版本的 Schema但问题没解决。后来我仔细对比了 官方文档里的示例Schema 和我生成的 Schema发现了两个问题。第一个问题我的 Schema 里有些字段用了oneOf和anyOf这是 JSON Schema 的特性但 AI平台的Action解析器不支持这些。它只认最基本的type、properties、required这几个字段。第二个问题更隐蔽。我的工具参数里有个字段叫input这跟 平台方 的保留字段冲突了。AI平台解析到这个字段名的时候直接报错但错误信息特别模糊只说 “Invalid schema”。解决办法是把所有参数名检查一遍避开平台方的保留字。同时把 Schema 里的oneOf、anyOf、$ref这些高级特性全部展开成最基本的类型定义。我写了一个 Schema 清洗函数来处理这个问题。defclean_openapi_schema(schema:dict)-dict: 清洗 OpenAPI Schema移除 AI平台不兼容的特性 确保 Schema 能被自定义助手正确解析 importcopy# 深拷贝避免修改原始数据cleanedcopy.deepcopy(schema)defclean_node(node:Any)-Any:递归清洗 Schema 节点ifisinstance(node,dict):# 移除 AI对话平台 不支持的字段forkeyin[oneOf,anyOf,allOf,$ref,discriminator]:node.pop(key,None)# 递归处理所有子节点forkey,valueinnode.items():node[key]clean_node(value)elifisinstance(node,list):# 列表中的每个元素也要递归清洗return[clean_node(item)foriteminnode]returnnode# 强制设置 OpenAPI 版本为 3.0.0cleaned[openapi]3.0.0returnclean_node(cleaned)# 在 /openapi.json 端点中使用清洗后的 Schemaapp.get(/openapi.json)asyncdefget_openapi_schema():返回清洗后的 OpenAPI Schemaschemaapp.openapi()# 添加工具端点定义同前面的代码fortoolincached_tools:# ... 端点定义代码 ...pass# 清洗后返回returnclean_openapi_schema(schema)加了这层清洗之后Schema 就能被 AI对话平台 正确导入了。这个坑的教训是不要想当然地认为标准 OpenAPI Schema 就能用AI对话平台 的 Action 解析器对 Schema 的支持远没有标准 OpenAPI 工具那么完善。还有一个相关的坑AI平台对Action的响应体大小有限制。如果你的 MCP 工具返回的内容特别大比如一个包含几千行日志的查询结果AI平台可能会截断或者直接报错。解决办法是在桥接服务里加一层内容截断逻辑超过一定长度就截断并附加提示信息。# 在 call_tool 路由中添加截断逻辑MAX_RESPONSE_LENGTH3000# 自定义助手Action响应最大字符数app.post(/tools/{tool_name})asyncdefcall_tool(tool_name:str,request:ToolCallRequest):# ... 前面的调用逻辑不变 ...result_text\n.join(texts)# 如果结果太长截断并提示iflen(result_text)MAX_RESPONSE_LENGTH:result_text(result_text[:MAX_RESPONSE_LENGTH]\n... [结果已截断完整结果请缩小查询范围])return{result:result_text}限制和解决方案用桥接方案接AI对话平台还有一些限制。第一是认证问题MCP Server 需要认证的话在桥接服务里处理环境变量配 token 或 自定义助手的Action设置里配API Key。第二是超时问题AI平台的Action调用有 30 秒超时限制超时了在桥接服务里返回友好提示。第三是并发问题单个 MCP Session 多用户同时调用可能冲突解决方案是维护 Session 池或给每个请求创建独立连接。小结这篇我们聊了怎么通过桥接让AI对话平台用上MCP。核心思路是写 FastAPI 服务把 MCP 的 stdio 协议转换成 HTTP REST API再通过 OpenAPI Schema 导入到自定义助手的Actions里。几个关键点。第一用 MCP Python SDK 的stdio_client连接 Server用 FastAPI 暴露端点。第二OpenAPI Schema 需要清洗去掉oneOf、anyOf等不支持的特性避开保留字。第三桥接服务必须部署到公网。第四注意响应大小和超时限制。虽然 平台方后来推出了原生MCP连接器但目前只支持审核过的远程 Server本地 Server 还是得靠桥接。这个方案在我们团队用了大半年稳定运行没出过大问题。下一篇我们看看 Windsurf 是怎么支持 MCP 的。相关推荐VS Code Copilot集成MCP与GitHub CopilotWindsurf集成AI编程新贵的MCP支持Completions与通知机制自动补全、进度上报与状态通知