企业官网建设流程全解析

1. 从一次客服工单说起：为什么要远程调用 MCP 服务

上周帮朋友看一个客服系统的工单，用户问「今天广州有没有三星级酒店能开发票」，机器人先查天气、再查酒店、最后还要确认发票政策，三个工具串起来才给出答案。这套链路背后其实就是 MCP（Model 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api ，注意 API 地址不带 UTM 参数。

你需要先拿到 Key，去控制台的 API Keys 页面创建：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。创建时建议按用途分 Key，比如「客服 Agent 专用」「内部工具专用」，方便后面按 Key 做限流和审计。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面写了 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(level=logging.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_url="https://taotoken.net/api", api_key=os.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(f"registered 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 f"data: {json.dumps({'type': 'start'})}\n\n" await asyncio.sleep(0) while True: response = self.client.chat.completions.create( model=self.model, messages=messages, tools=self.tools, temperature=0, max_tokens=16000, stream=True, ) 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 f"data: {json.dumps({'type':'final_content','content':delta.content}, ensure_ascii=False)}\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 f"data: {json.dumps({'type':'end','content':collected_content}, ensure_ascii=False)}\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 f"data: {json.dumps({'type':'tool_executing','tool':tool_name}, ensure_ascii=False)}\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 用 .text，2.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 f"data: {json.dumps({'type':'tool_result','tool':tool_name,'result':tool_result}, ensure_ascii=False)}\n\n" await asyncio.sleep(0) except Exception as e: messages.append({ 'role': 'tool', 'tool_call_id': tool_id, 'content': f"Error executing tool: {str(e)}", }) yield f"data: {json.dumps({'type':'tool_error','tool':tool_name,'error':str(e)}, ensure_ascii=False)}\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(title="Remote MCP Client API", version="1.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_code=500, detail="MCP Client not initialized") async def generate(): messages = limit_messages(request.messages, max_rounds=5) 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_type="text/event-stream") @app.get("/tools") async def get_tools(): if not mcp_client: raise HTTPException(status_code=500, detail="MCP Client not initialized") return {"tools": mcp_client.tools} @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == '__main__': uvicorn.run(app, host="0.0.0.0", port=8000)

场景 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 逐行读 SSE：

import 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, json=payload, headers=headers, stream=True) as resp: for line in resp.iter_lines(decode_unicode=True): 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].text，2.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是否清晰。描述太模糊模型会忽略工具。另外temperature=0能提高工具调用的稳定性。

排障时如果怀疑是 Key 或通道问题，去 API Keys 页面确认 Key 状态：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，接入细节看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。想先验证模型本身通不通，可以用模型对话页面发一条消息试试：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite 。

7. 长期编码与 Agent 场景的通道选择

如果你只是偶尔跑一次远程 MCP 调用，按上面的配置用 API Key 就够了。但如果你在做长期的 Agent 项目，每天要跑几十上百次工具调用，建议看看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。它更适合持续性的编码和 Agent 任务，配额和计费方式对高频调用更友好。

回到这套架构本身，核心就三件事：统一 Key 收敛鉴权、settings.json集中注册工具、fastapi 暴露 SSE 接口。工具增减只改配置不改代码，客户端和服务端解耦。我试过把工具从 3 个扩到 12 个，客户端一行没动，只在settings.json里加了几个 url。这套骨架你拿去改改就能用，先把/health和/tools跑通，再调/chat/stream，链路就活了。