1. 从一次客服工单说起为什么要远程调用 MCP 服务上周帮朋友看一个客服系统的工单用户问「今天广州有没有三星级酒店能开发票」机器人先查天气、再查酒店、最后还要确认发票政策三个工具串起来才给出答案。这套链路背后其实就是 MCPModel Context Protocol在干活大模型负责判断该调哪个工具客户端负责真正把工具跑起来再把结果喂回模型继续推理。问题在于很多团队把 MCP 服务写成了本地脚本工具一多就散落在各个机器上客服端、运营端、内部系统各连各的Key 也各管各的。fastmcp 客户端远程 MCP 服务调用要解决的就是这件事用一个统一的 Key/API 通道把多工具 MCP 服务集中注册客户端通过 HTTP 远程调用再以 fastapi 集成的方式对外暴露 SSE 流式接口。适合谁正在做 Agent 工具链、客服机器人、内部自动化平台的开发者尤其是工具数量超过 3 个、需要跨服务复用的场景。这篇会给你三样东西一份可复制的 config.toml / settings.json 配置骨架一段多工具注册与调用的 fastapi 集成代码以及远程连通性验证的具体动作。版本上要注意fastmcp 2.8.0 里call_tool的结果直接.text就能拿到而 2.11.3 之后改成了.content这个差异后面排障会专门讲。2. TaoToken 前置统一 Key 与 API 通道怎么接远程 MCP 服务调用最烦的是鉴权分散。我的做法是把模型调用和工具调用都收敛到同一个 API 通道上客户端只认一个 Key工具服务注册在统一入口后面。TaoToken 在这里扮演的就是这个统一通道的角色官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。你需要先拿到 Key去控制台的 API Keys 页面创建https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。创建时建议按用途分 Key比如「客服 Agent 专用」「内部工具专用」方便后面按 Key 做限流和审计。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面写了 base_url 和兼容 OpenAI SDK 的调用方式。这里要强调一点TaoToken 是合规的 API 聚合通道不是灰色中转所有调用都走标准 HTTPS客户端配置里只需要填 base_url 和 api_key 两个字段。如果你之前用的是 dashscope 的 compatible-mode把 base_url 换成 TaoToken 的 API 地址即可OpenAI SDK 的代码几乎不用改。3. 可复制配置config.toml 与 settings.json 骨架配置分两层一层是客户端侧的模型与通道配置一层是 MCP 服务侧的注册配置。先看客户端侧我习惯用config.toml管理放在项目根目录# config.toml [llm] base_url https://taotoken.net/api api_key sk-你的Key model qwen-plus temperature 0 max_tokens 16000 [mcp] # 远程 MCP 服务地址多个服务用数组 servers [ https://your-mcp-host:7080/mcp, ] # 单次会话最大轮数超过则截断历史 max_rounds 5 # 工具调用超时秒 tool_timeout 30 [server] host 0.0.0.0 port 8000再看 MCP 服务侧的settings.json用于声明工具注册信息。如果你用的是 fastmcp 的 server 端可以这样写{ mcpServers: { hotel: { url: https://your-mcp-host:7080/mcp, tools: [search_hotel, check_invoice], timeout: 30 }, weather: { url: https://your-mcp-host:7081/mcp, tools: [get_weather], timeout: 15 }, flight: { url: https://your-mcp-host:7082/mcp, tools: [search_flight], timeout: 30 } } }两个配置的分工要清楚config.toml管的是「客户端怎么连模型、连哪些 MCP 服务」settings.json管的是「每个 MCP 服务暴露了哪些工具、超时多少」。实际部署时settings.json可以放在服务端客户端只读config.toml这样工具增减不用改客户端代码。注意api_key不要硬编码进代码仓库用环境变量TAOTOKEN_API_KEY注入代码里读os.getenv。4. fastapi 集成多工具注册与调用示例下面这段是核心把 MCP 客户端封装成类再挂到 fastapi 上。先看 MCPClient 的封装重点是prepare_tools把远程工具列表转成 OpenAI function calling 格式#!/usr/bin/python3 # -*- coding: utf-8 -*- import json import asyncio import os import logging from typing import List, Dict, Optional from openai import OpenAI from fastmcp import Client logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s) class MCPClient: def __init__(self, script: str, model: str qwen-plus): self.script script self.model model self.client OpenAI( base_urlhttps://taotoken.net/api, api_keyos.getenv(TAOTOKEN_API_KEY), ) self.session Client(script) self.tools [] self._connected False async def connect(self): if not self._connected: await self.session.__aenter__() self._connected True async def disconnect(self): if self._connected: await self.session.__aexit__(None, None, None) self._connected False async def prepare_tools(self): if not self.tools: await self.connect() tools await self.session.list_tools() self.tools [ { type: function, function: { name: tool.name, description: tool.description, parameters: tool.inputSchema, }, } for tool in tools ] logging.info(fregistered tools: {[t[function][name] for t in self.tools]})多工具注册的关键在list_tools()返回的inputSchema它直接就是 JSON Schema塞进 function calling 的parameters字段即可不用手动转换。接下来是流式对话逻辑支持多轮和多工具串联async def chat_stream(self, messages: List[Dict]): if not self.tools: await self.prepare_tools() yield fdata: {json.dumps({type: start})}\n\n await asyncio.sleep(0) while True: response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.tools, temperature0, max_tokens16000, streamTrue, ) collected_content collected_tool_calls [] finish_reason None for chunk in response: if not chunk.choices: continue delta chunk.choices[0].delta finish_reason chunk.choices[0].finish_reason if delta.content: collected_content delta.content yield fdata: {json.dumps({type:final_content,content:delta.content}, ensure_asciiFalse)}\n\n await asyncio.sleep(0) if hasattr(delta, tool_calls) and delta.tool_calls: if not collected_tool_calls: collected_tool_calls [ {id: , function: {name: , arguments: }} for _ in delta.tool_calls ] for i, tc in enumerate(delta.tool_calls): if tc.id: collected_tool_calls[i][id] tc.id if tc.function: if tc.function.name: collected_tool_calls[i][function][name] tc.function.name if tc.function.arguments: collected_tool_calls[i][function][arguments] tc.function.arguments if finish_reason ! tool_calls or not collected_tool_calls: yield fdata: {json.dumps({type:end,content:collected_content}, ensure_asciiFalse)}\n\n break messages.append({ role: assistant, content: collected_content, tool_calls: [ { id: tc[id], type: function, function: { name: tc[function][name], arguments: tc[function][arguments], }, } for tc in collected_tool_calls ], }) for tool_call in collected_tool_calls: tool_name tool_call[function][name] tool_args_str tool_call[function][arguments] tool_id tool_call[id] if not tool_name: continue yield fdata: {json.dumps({type:tool_executing,tool:tool_name}, ensure_asciiFalse)}\n\n try: args json.loads(tool_args_str) if tool_args_str else {} result await self.session.call_tool(tool_name, args) # fastmcp 2.8.0 用 .text2.11.3 用 .content tool_result result[0].text if result else No result messages.append({ role: tool, tool_call_id: tool_id, content: tool_result, }) yield fdata: {json.dumps({type:tool_result,tool:tool_name,result:tool_result}, ensure_asciiFalse)}\n\n await asyncio.sleep(0) except Exception as e: messages.append({ role: tool, tool_call_id: tool_id, content: fError executing tool: {str(e)}, }) yield fdata: {json.dumps({type:tool_error,tool:tool_name,error:str(e)}, ensure_asciiFalse)}\n\n await asyncio.sleep(0)注意call_tool那行result[0].text是 2.8.0 的写法。如果你升级到 2.11.3要改成result[0].content否则会报AttributeError。这是版本差异里最容易踩的坑。最后挂到 fastapi加上会话轮数限制和场景 prompt 拼接from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel import uvicorn BASE_SYSTEM_PROMPT 你是智能助手能够调用各种MCP工具帮助用户高效解决各类问题。 请根据用户需求合理选择和调用工具结果简洁明了。如遇信息不全可主动提问补全。 SCENE_PROMPTS { 出差行程规划: 在出差行程规划场景下请优先按顺序调用天气、酒店、机票等工具最后汇总回复。, 报销流程: 在报销场景下请指导用户如何填写报销单并可调用相关工具自动生成报销明细。, } def build_system_prompt(scene: str None): if scene and scene in SCENE_PROMPTS: return BASE_SYSTEM_PROMPT \n SCENE_PROMPTS[scene] return BASE_SYSTEM_PROMPT def limit_messages(messages: List[Dict], max_rounds: int 5) - List[Dict]: system_msg [m for m in messages if m[role] system][:1] if not system_msg: system_msg [{role: system, content: build_system_prompt()}] others [m for m in messages if m[role] ! system] if len(others) max_rounds * 2: others others[-max_rounds * 2:] return system_msg others class ChatRequest(BaseModel): messages: List[Dict] scene: Optional[str] None stream: bool True app FastAPI(titleRemote MCP Client API, version1.0.0) mcp_client None app.on_event(startup) async def startup_event(): global mcp_client mcp_client MCPClient(https://your-mcp-host:7080/mcp) await mcp_client.prepare_tools() app.on_event(shutdown) async def shutdown_event(): global mcp_client if mcp_client: await mcp_client.disconnect() app.post(/chat/stream) async def chat_stream_endpoint(request: ChatRequest): if not mcp_client: raise HTTPException(status_code500, detailMCP Client not initialized) async def generate(): messages limit_messages(request.messages, max_rounds5) if request.scene: messages[0][content] build_system_prompt(request.scene) async for chunk in mcp_client.chat_stream(messages): yield chunk return StreamingResponse(generate(), media_typetext/event-stream) app.get(/tools) async def get_tools(): if not mcp_client: raise HTTPException(status_code500, detailMCP Client not initialized) return {tools: mcp_client.tools} app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)场景 prompt 的拼接逻辑是主 prompt 描述通用能力场景补充 prompt 按需动态拼。用户意图识别出「出差」就传scene出差行程规划工具调用顺序会被引导成天气→酒店→机票最后汇总。5. 验证请求从 curl 到 Python 客户端服务起来后先跑健康检查确认 fastapi 和 MCP 连接都正常curl http://localhost:8000/health # {status:healthy} curl http://localhost:8000/tools # {tools:[{type:function,function:{name:search_hotel,...}}]}/tools返回的列表就是远程 MCP 服务注册上来的工具如果这里是空的说明prepare_tools没跑通回去看 MCP 服务地址和网络连通性。单轮对话用 curl 验证 SSE 流curl -N -X POST http://localhost:8000/chat/stream \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d {messages:[{role:user,content:现在时间}]}你会看到data: {type:start}、data: {type:tool_executing,tool:get_time}、data: {type:tool_result,...}、data: {type:end,...}依次返回。多轮对话把历史消息一起传curl -X POST http://localhost:8000/chat/stream \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d {messages:[ {role:user,content:推荐下酒店}, {role:assistant,content:请提供城市、日期、星级要求}, {role:user,content:今天 广州 3星级} ],scene:出差行程规划}Python 客户端验证更直观用 requests 逐行读 SSEimport requests import json url http://localhost:8000/chat/stream payload { messages: [{role: user, content: 今天广州天气怎么样}], scene: 出差行程规划, } headers {Content-Type: application/json, Accept: text/event-stream} with requests.post(url, jsonpayload, headersheaders, streamTrue) as resp: for line in resp.iter_lines(decode_unicodeTrue): if line and line.startswith(data: ): data json.loads(line[6:]) print(data)实测下来多工具串联时事件顺序是start→tool_executing(weather)→tool_result(weather)→tool_executing(hotel)→tool_result(hotel)→final_content→end。前端拿到tool_executing就可以显示「正在查询天气…」体验比等最终结果好很多。6. 本篇常见错排查报错一AttributeError: CallToolResult object has no attribute text这是版本差异。fastmcp 2.8.0 用result[0].text2.11.3 改成result[0].content。先pip show fastmcp看版本再决定用哪个属性。想兼容两个版本可以写item result[0] tool_result getattr(item, text, None) or getattr(item, content, )报错二/tools返回空列表大概率是 MCP 服务地址不通。先用curl https://your-mcp-host:7080/mcp看能不能握手再检查settings.json里的 url 是否和MCPClient初始化时传的 script 一致。另外注意prepare_tools是异步的如果在startup_event里没 await工具列表会来不及加载。报错三SSE 流中断前端收不到end事件检查chat_stream里的while True循环finish_reason ! tool_calls时必须 break 并 yield end。如果工具调用后模型继续返回内容循环会再跑一轮这是正常的 Agentic Loop。但如果工具一直返回错误模型可能反复调用同一个工具建议在limit_messages里限制轮数或者在工具错误时直接终止。报错四多轮对话历史丢失limit_messages默认保留最近 5 轮10 条消息超过就截断。如果你需要更长记忆把max_rounds调大但注意 token 消耗。传空列表messages: []会清空会话这是设计好的行为。报错五模型不调用工具直接回答检查tools参数是否传进了chat.completions.create以及工具的description是否清晰。描述太模糊模型会忽略工具。另外temperature0能提高工具调用的稳定性。排障时如果怀疑是 Key 或通道问题去 API Keys 页面确认 Key 状态https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。想先验证模型本身通不通可以用模型对话页面发一条消息试试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 。7. 长期编码与 Agent 场景的通道选择如果你只是偶尔跑一次远程 MCP 调用按上面的配置用 API Key 就够了。但如果你在做长期的 Agent 项目每天要跑几十上百次工具调用建议看看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。它更适合持续性的编码和 Agent 任务配额和计费方式对高频调用更友好。回到这套架构本身核心就三件事统一 Key 收敛鉴权、settings.json集中注册工具、fastapi 暴露 SSE 接口。工具增减只改配置不改代码客户端和服务端解耦。我试过把工具从 3 个扩到 12 个客户端一行没动只在settings.json里加了几个 url。这套骨架你拿去改改就能用先把/health和/tools跑通再调/chat/stream链路就活了。