MCP服务器从零搭建：基于HTTP流式传输的FastAPI实现与TaoToken接入

1. 从零搭建 MCP 服务器为什么选 FastAPI HTTP 流式传输MCPModel Context Protocol服务器本质上是一个“工具插座”大模型通过它调用外部能力比如查天气、读数据库、发消息。传统做法多用 stdio 传输客户端和服务器必须跑在同一台机器、同一个进程组里一旦想放到远端或容器里就非常别扭。HTTP 流式传输的 MCP 服务器解决的正是这个问题——服务器可以独立部署客户端通过标准 HTTP 请求接入工具调用的中间进度还能以流的方式实时吐回来。这套方案适合谁如果你正在做 AI Agent、想让本地或远端的大模型调用自定义工具又不想被 stdio 的进程绑定限制那 FastAPI HTTP 流式传输就是很顺手的组合。FastAPI 自带异步、类型校验和自动文档写 MCP 的 JSON-RPC 路由非常省事流式响应则用StreamingResponse配合异步生成器几行代码就能把工具执行过程分块推给客户端。我试过用纯 stdio 写 MCP联调时客户端一崩服务器就跟着挂日志还混在一起。换成 HTTP 之后服务器可以单独用uvicorn跑着客户端崩了重连就行排查也清晰。下面我会从项目初始化开始一步步给出可复制的 FastAPI 路由、MCP 工具注册表、流式响应实现再用 curl 和自写客户端完成一次完整调用验证最后把模型通道接到 TaoToken 上让整个链路跑通。核心检索词先明确MCP 服务器、HTTP 流式传输、FastAPI、客户端接入。这四个词会贯穿全文你跟着做就能得到一个能实际调用的服务。2. TaoToken 前置准备统一 Key 与 API 通道在写客户端之前先把模型通道准备好。MCP 服务器负责“执行工具”但真正决定要不要调工具的是大模型所以客户端里需要一个能走 Function Calling 的模型接口。TaoToken 在这里的作用是提供统一的 Key 和 API 通道你不用为每个模型单独配一套鉴权和地址。你需要准备两样东西一个 API Key以及确认要用的模型 ID。Key 在控制台创建地址是https://taotoken.net/api-keys登录后新建即可。模型 ID 则根据你实际要用的模型填比如做工具调用建议选支持 Function Calling 的模型。Base URL 统一用https://taotoken.net/api注意这个地址不带任何查询参数。把这三件套记下来后面写.env和客户端配置时会直接用到配置项值说明Base URLhttps://taotoken.net/api所有请求的统一入口API Key控制台创建放在.env不要硬编码Model ID按需选择需支持 Function Calling注意API Key 只放在服务端或本地.env文件里不要提交到代码仓库也不要在前端明文暴露。如果你还没创建 Key可以先打开https://taotoken.net/api-keys建一个。想先验证模型通道是否正常可以用模型对话页面发一条测试消息确认返回正常再继续。对于长期做编码或 Agent 的场景Coding Plan 会更省心地址是https://taotoken.net/coding-plan。这一步不涉及任何服务器代码但它是后面客户端能跑通的前提。很多人卡在“工具调用了但模型没反应”最后发现是 Key 或 Base URL 配错所以先把这块确认清楚。3. 可复制配置FastAPI 路由与 MCP 工具注册现在进入正题。先初始化项目我用uv管理依赖你也可以用 pip命令等价。uv init mcp-weather-http cd mcp-weather-http uv venv source .venv/bin/activate uv add mcp httpx fastapi uvicorn python-dotenv openai mkdir -p ./src/mcp_weather_http cd ./src/mcp_weather_http接着创建server.py。这个文件实现三个核心能力initialize能力协商、tools/list工具注册、tools/call流式执行。先看工具注册表它决定了模型能看到哪些工具TOOLS_REGISTRY { tools: [ { name: get_weather, description: 查询指定城市的当前天气输入城市英文名称。, inputSchema: { type: object, properties: { city: { type: string, description: City name, e.g. Hangzhou } }, required: [city] } } ], nextCursor: None }inputSchema用的是 JSON Schema模型据此生成参数。nextCursor为None表示工具列表不分页一次返回完。然后是 FastAPI 路由。MCP 的 JSON-RPC 方法都走POST /mcpGET /mcp用于客户端探测from fastapi import FastAPI, Request, Response, status from fastapi.responses import StreamingResponse app FastAPI(titleWeatherServer HTTP-Stream) PROTOCOL_VERSION 2024-11-05 app.get(/mcp) async def mcp_probe(): return { jsonrpc: 2.0, id: 0, result: { protocolVersion: PROTOCOL_VERSION, capabilities: {streaming: True, tools: {listChanged: True}}, serverInfo: {name: WeatherServer, version: 1.0.0}, instructions: Use get_weather to fetch weather by city name. } } app.post(/mcp) async def mcp_endpoint(request: Request): body await request.json() req_id body.get(id, 1) method body.get(method) if method notifications/initialized: return Response(status_codestatus.HTTP_204_NO_CONTENT) if method initialize: return { jsonrpc: 2.0, id: req_id, result: { protocolVersion: PROTOCOL_VERSION, capabilities: {streaming: True, tools: {listChanged: True}}, serverInfo: {name: WeatherServer, version: 1.0.0} } } if method tools/list: return {jsonrpc: 2.0, id: req_id, result: TOOLS_REGISTRY} if method tools/call: params body.get(params, {}) city params.get(arguments, {}).get(city) if not city: return {jsonrpc: 2.0, id: req_id, error: {code: -32602, message: Missing city}} return StreamingResponse(stream_weather(city, req_id), media_typeapplication/json) return {jsonrpc: 2.0, id: req_id, error: {code: -32601, message: Method not found}}流式响应的关键在stream_weather它是一个异步生成器先吐一条进度再吐最终结果import asyncio, json from typing import AsyncIterator async def stream_weather(city: str, req_id) - AsyncIterator[bytes]: yield json.dumps({ jsonrpc: 2.0, id: req_id, stream: f查询 {city} 天气中… }).encode() b\n await asyncio.sleep(0.3) data await fetch_weather(city) if error in data: yield json.dumps({ jsonrpc: 2.0, id: req_id, error: {code: -32000, message: data[error]} }).encode() b\n return yield json.dumps({ jsonrpc: 2.0, id: req_id, result: { content: [{type: text, text: format_weather(data)}], isError: False } }).encode() b\nfetch_weather用httpx.AsyncClient请求天气接口format_weather把 JSON 转成可读文本。启动入口用 argparse 接收 API Key 和端口def main(): import argparse, uvicorn parser argparse.ArgumentParser() parser.add_argument(--api_key, requiredTrue) parser.add_argument(--host, default127.0.0.1) parser.add_argument(--port, typeint, default8000) args parser.parse_args() global API_KEY API_KEY args.api_key uvicorn.run(app, hostargs.host, portargs.port, log_levelinfo)启动命令uv run ./src/mcp_weather_http/server.py --api_key YOUR_WEATHER_KEY到这里一个支持 HTTP 流式传输的 MCP 服务器就成型了。工具注册、能力协商、流式执行三块都齐了。4. 验证请求curl 与客户端联调成功结果服务器跑起来后先用 curl 模拟 MCP 客户端的标准流程确认每一步返回符合预期。第一步initialize能力协商curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05}}期望返回protocolVersion、capabilities和serverInfo。如果这里报错说明路由或 JSON 解析有问题。第二步发送notifications/initialized通知确认上线curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:notifications/initialized}期望返回 204没有响应体。这是通知类消息不需要回复。第三步tools/list获取工具注册表curl -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/list,params:{}}期望返回get_weather的完整 schema。这一步验证工具注册是否正确。第四步tools/call流式调用注意加-N关闭缓冲curl -N -X POST http://localhost:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:3,method:tools/call,params:{name:get_weather,arguments:{city:Hangzhou}}}你会先看到一条stream进度再看到result.content里的天气文本。这就是 HTTP 流式传输的效果——中间进度和最终结果分块到达。curl 验证通过后写一个自包含的客户端把模型接进来。创建client.py核心是HTTPMCPServer类它封装了 initialize、list_tools 和 call_tool_streamimport httpx, json, os from openai import OpenAI from dotenv import load_dotenv class HTTPMCPServer: def __init__(self, name, endpoint): self.name name self.endpoint endpoint.rstrip(/) self.session None async def initialize(self): self.session httpx.AsyncClient(timeout30.0) await self._post_json({ jsonrpc: 2.0, id: 0, method: initialize, params: {protocolVersion: 2024-11-05, capabilities: {}, clientInfo: {name: HTTP-MCP-Demo, version: 0.1}} }) await self._post_json({jsonrpc: 2.0, method: notifications/initialized}) async def list_tools(self): res await self._post_json({jsonrpc: 2.0, id: 1, method: tools/list, params: {}}) return res[result][tools] async def call_tool_stream(self, tool_name, arguments): req {jsonrpc: 2.0, id: 3, method: tools/call, params: {name: tool_name, arguments: arguments}} collected [] async with self.session.stream(POST, self.endpoint, jsonreq, headers{Accept: application/json}) as resp: async for line in resp.aiter_lines(): if not line: continue chunk json.loads(line) if stream in chunk: continue if result in chunk: for item in chunk[result][content]: if item[type] text: collected.append(item[text]) return \n.join(collected) async def _post_json(self, payload): r await self.session.post(self.endpoint, jsonpayload, headers{Accept: application/json}) if r.status_code 204 or not r.content: return {} r.raise_for_status() return r.json()模型侧用 OpenAI SDK 指向 TaoToken 的 Base URL。.env文件这样写LLM_API_KEY你的TaoToken_Key BASE_URLhttps://taotoken.net/api MODEL你的模型IDservers_config.json记录服务器地址{ mcpServers: { weather: { endpoint: http://127.0.0.1:8000/mcp } } }主循环里模型返回tool_calls时解析出工具名和参数调用call_tool_stream把结果作为tool消息回填再请求一次模型生成最终回答。启动客户端uv run ./src/mcp_weather_http/client.py输入“杭州天气怎么样”你会看到[调用工具] weather_get_weather → {city: Hangzhou}然后模型基于天气文本给出自然语言回答。整条链路——MCP 服务器、HTTP 流式传输、TaoToken 模型通道——就完整跑通了。5. 本篇常见错排查401、local proxy failed 与 reading choices联调时最容易撞的几个报错我按实际遇到的频率列出来对照着改。401 Unauthorized。这个几乎都是 Key 的问题。先确认.env里LLM_API_KEY没有多余空格或引号再确认BASE_URL是https://taotoken.net/api不要多加/v1或斜杠。如果 Key 是在控制台刚创建的确认没有复制错位。还有一种情况是 Key 被禁用或额度耗尽去控制台看一眼状态。local proxy failed / connection refused。客户端报这个通常是 MCP 服务器没启动或者servers_config.json里的 endpoint 端口写错。先curl http://127.0.0.1:8000/mcp确认服务器活着。如果服务器在容器里注意127.0.0.1在容器内指向容器自身要用宿主 IP 或容器网络别名。reading choices 报错。这个出现在模型返回结构不符合预期时常见原因是模型不支持 Function Calling或者tools参数格式不对。检查MODEL是否选了支持工具调用的模型再检查all_tools里每个工具的parameters是否直接用了inputSchema字段名必须是parameters不能写成input_schema。OAuth / 鉴权相关报错。如果你用的是需要 OAuth 的客户端注意 MCP 服务器本身不做 OAuth鉴权在模型通道那层。确认 TaoToken 的 Key 是通过Authorization: Bearer传递的OpenAI SDK 会自动处理。如果手动拼请求别漏了Bearer前缀。流式响应收不到中间进度。curl 不加-N会缓冲客户端用httpx的stream方法时aiter_lines要配合async for。如果只收到最终结果没有进度检查stream_weather里第一条yield是否真的执行了以及media_type是否为application/json。tools/list 返回空。检查TOOLS_REGISTRY的tools数组是否为空以及tools/list分支是否真的返回了它。有时候是method字符串拼错比如写成tool/list。把这几条对照一遍基本能覆盖 90% 的联调问题。剩下 10% 看服务器日志uvicorn会把每个请求的 method 打出来定位很快。6. 语义一致 CTA把链路接到 TaoToken整套流程跑通后你会发现 MCP 服务器负责工具执行模型通道负责决策两者通过 HTTP 解耦。TaoToken 在这里承担的是统一 Key 和 API 通道的角色让你不用为每个模型单独维护鉴权。如果你还在配 Key 阶段直接去https://taotoken.net/api-keys创建然后按第 3 节的.env格式填进去。想先确认模型通道正常用模型对话页面发一条消息试试地址是https://taotoken.net/model-chat。接入文档在https://taotoken.net/doc里面有 Base URL、鉴权和 Function Calling 的完整说明。对于长期跑编码或 Agent 的场景Coding Plan 比按量更划算地址是https://taotoken.net/coding-plan。控制台在https://taotoken.net/console可以看用量和 Key 状态。最后给一个实用技巧把 MCP 服务器的启动命令和客户端启动命令写成两个 shell 脚本联调时分别开两个终端跑日志互不干扰。服务器端日志看uvicorn的请求 method客户端日志看工具调用参数和模型返回两边一对问题基本无处藏身。