企业官网建设流程全解析

1. 从零搭建 MCP 服务器：为什么选 FastAPI + HTTP 流式传输

MCP（Model 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和客户端配置时会直接用到：

注意：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 /mcp，GET /mcp用于客户端探测：

from fastapi import FastAPI, Request, Response, status from fastapi.responses import StreamingResponse app = FastAPI(title="WeatherServer 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_code=status.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_type="application/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"\n"

fetch_weather用httpx.AsyncClient请求天气接口，format_weather把 JSON 转成可读文本。启动入口用 argparse 接收 API Key 和端口：

def main(): import argparse, uvicorn parser = argparse.ArgumentParser() parser.add_argument("--api_key", required=True) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8000) args = parser.parse_args() global API_KEY API_KEY = args.api_key uvicorn.run(app, host=args.host, port=args.port, log_level="info")

启动命令：

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_stream：

import 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(timeout=30.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, json=req, 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, json=payload, 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_URL=https://taotoken.net/api MODEL=你的模型ID

servers_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，客户端日志看工具调用参数和模型返回，两边一对，问题基本无处藏身。