1. 从零跑通大模型通信链路:FastAPI + SSE MCP 服务到底解决什么问题
如果你正在做 AI Agent 或者智能硬件后端,大概率会遇到一个很具体的场景:你有一堆现成的 Python 业务接口(查天气、查订单、读传感器数据),想让大模型直接调用它们,但每次对接都要手写一套工具描述、参数校验、流式返回,改一个字段就要动三四个文件。MCP(Model Context Protocol)就是为了解决这个"工具接入标准化"的问题而出现的,而 FastAPI + SSE 的组合,是目前 Python 技术栈里落地成本最低的一条路。
这篇文章要讲清楚三件事:第一,用 FastAPI 把普通 HTTP 接口自动暴露成 MCP 工具;第二,通过 TaoToken 的统一 Key 和 API 通道调用大模型,不用在代码里散落多个厂商的 Key;第三,交付一份可以直接复制的config.toml骨架、uvicorn 启动命令,以及一次真实的 SSE 流式请求验证动作。目标很明确——从零把大模型通信链路跑通,而不是停留在概念介绍。
适合谁看:有 Python 基础、写过 FastAPI 或 Flask 接口、想快速把业务能力接进大模型工具链的开发者;也适合正在做智能硬件网关、需要让设备侧通过统一通道调用大模型的同学。全文的代码都可以直接跑,配置项我会标注清楚哪些必须改、哪些保持默认即可。
先说结论性的架构判断:SSE 是单向的服务端推送,MCP 客户端到服务端仍然走 HTTP POST,所以整体是"POST 上行 + SSE 下行"的伪双工模式。这一点决定了你在 FastAPI 里不能只写一个 GET 接口就完事,需要理解请求和响应是分开的两条通道。理解了这一点,后面的配置和排障都会顺很多。
2. TaoToken 前置准备:统一 Key 与 API 通道怎么配
在写 MCP 服务之前,先把大模型调用通道准备好。TaoToken 的作用是把多家模型的调用收敛到一个 Base URL 和一把 Key 上,这样你的 MCP 服务里只需要维护一份凭证,换模型时改一个 Model ID 就行,不用去动业务代码。
你需要准备三样东西,我把它叫做"三件套":Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 OpenAI 兼容接口的根路径使用。API Key 在控制台的 API Keys 页面创建,创建后只显示一次,建议立刻复制到环境变量里,不要硬编码进代码。
具体操作路径是这样的:先访问官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 了解整体能力,然后进入控制台 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 创建 Key,再到 API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 管理你的密钥。如果你只是想先验证模型能不能通,可以直接用模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 发一条消息试试,确认通道没问题再写代码。
环境变量建议这样设置,Linux/macOS 用 export,Windows 用 set:
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL_ID="你的模型ID"这里有个容易踩的坑:Base URL 结尾不要多加/v1或者斜杠。很多 OpenAI 兼容客户端会自动拼接/v1/chat/completions,如果你手动加了/v1,最终路径会变成/v1/v1/chat/completions,直接 404。我实测下来,保持https://taotoken.net/api这个形式最稳。
Model ID 的填写要和你在控制台看到的模型名称完全一致,大小写敏感。如果你不确定用哪个,先在模型对话页面选一个能正常回复的,把它的标识复制过来。对于长期做编码和 Agent 的场景,可以考虑 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ,它在高频调用下更划算;如果只是偶尔验证,按量调用即可。
把这三件套准备好之后,你的 MCP 服务就只需要读取环境变量,不需要在代码里出现任何明文密钥。这一步做完,后面的配置才有意义。
3. 可复制配置:config.toml 骨架与 FastAPI MCP 服务代码
这一节是全文的核心,我会给出完整的config.toml骨架和 FastAPI 服务代码,你复制过去改几个字段就能跑。
先看config.toml。这个文件放在项目根目录,用来集中管理服务端口、MCP 挂载路径和模型通道参数:
# config.toml - MCP 服务与大模型通道配置骨架 [server] host = "0.0.0.0" port = 8001 reload = true [mcp] mount_path = "/mcp" name = "Weather MCP Server" describe_all_responses = true describe_full_response_schema = true [llm] base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" model_id = "你的模型ID" timeout = 30.0 [upstream] nws_api_base = "https://api.weather.gov" user_agent = "weather-app/1.0"注意api_key_env这一项,它存的是环境变量的名字而不是 Key 本身,这样配置文件可以安全地提交到仓库。model_id需要你替换成实际值。
接下来是服务代码server.py。这里用fastapi-mcp把 FastAPI 端点自动转成 MCP 工具,同时保留原有的 HTTP 文档:
import os import tomllib import httpx from typing import Any from fastapi import FastAPI from fastapi_mcp import add_mcp_server with open("config.toml", "rb") as f: cfg = tomllib.load(f) app = FastAPI(title="MCP Gateway") mcp_server = add_mcp_server( app, mount_path=cfg["mcp"]["mount_path"], name=cfg["mcp"]["name"], describe_all_responses=cfg["mcp"]["describe_all_responses"], describe_full_response_schema=cfg["mcp"]["describe_full_response_schema"], ) NWS_API_BASE = cfg["upstream"]["nws_api_base"] USER_AGENT = cfg["upstream"]["user_agent"] async def make_nws_request(url: str) -> dict[str, Any] | None: headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"} async with httpx.AsyncClient() as client: try: resp = await client.get(url, headers=headers, timeout=cfg["llm"]["timeout"]) resp.raise_for_status() return resp.json() except Exception: return None @mcp_server.tool() async def get_forecast(latitude: float, longitude: float) -> str: """获取指定经纬度的天气预报。 参数: latitude: 纬度 longitude: 经度 """ points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}" points_data = await make_nws_request(points_url) if not points_data: return "无法获取该位置的预报数据。" forecast_url = points_data["properties"]["forecast"] forecast_data = await make_nws_request(forecast_url) if not forecast_data: return "无法获取详细预报。" periods = forecast_data["properties"]["periods"] return "\n---\n".join( f"{p['name']}: {p['temperature']}°{p['temperatureUnit']} " f"{p['windSpeed']} {p['windDirection']}\n{p['detailedForecast']}" for p in periods[:5] )安装依赖:
pip install fastapi uvicorn fastapi-mcp httpx启动服务:
uvicorn server:app --host 0.0.0.0 --port 8001 --reload启动后你会看到 uvicorn 输出监听地址,MCP 服务挂在http://127.0.0.1:8001/mcp。这里的关键点是add_mcp_server的mount_path参数,它决定了 SSE 端点的路径,客户端连接时必须和这个路径一致。
如果你用的是 Claude Code 这类工具,它的配置通常放在settings.json或项目级配置里,Base URL 填https://taotoken.net/api,Key 填环境变量引用,Model ID 填你在控制台选的模型。三件套缺一不可,尤其是 Model ID,漏填会直接报模型不存在。
4. 验证请求:一次真实的 SSE 流式调用与成功结果
服务起来之后,必须做一次真实的 SSE 请求验证,否则你不知道链路到底通没通。这里分两步:先验证 MCP 服务本身,再验证大模型通道。
第一步,用 MCP Inspector 连接 SSE 端点。启动 inspector:
CLIENT_PORT=8081 SERVER_PORT=8082 npx -y @modelcontextprotocol/inspector打开浏览器访问 inspector 提示的地址,在连接配置里选择 SSE 传输方式,URL 填http://127.0.0.1:8001/mcp。连接成功后,你应该能在工具列表里看到get_forecast,参数是 latitude 和 longitude。填入一组真实坐标,比如40.71和-74.01,点击调用,右侧会流式返回预报文本。看到分段的天气数据逐条出现,说明 SSE 下行通道正常。
第二步,验证大模型通道。用 curl 直接打 TaoToken 的兼容接口:
curl -N https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "'"$TAOTOKEN_MODEL_ID"'", "stream": true, "messages": [{"role": "user", "content": "用一句话说明 SSE 和 WebSocket 的区别"}] }'-N参数关闭 curl 的缓冲,这样你能看到data:开头的分块逐条打印出来。如果看到类似data: {"choices":[{"delta":{"content":"SSE"}}]}的输出,并且最后有data: [DONE],说明流式通道完全打通。
成功结果的判断标准有三个:HTTP 状态码 200、响应头里content-type是text/event-stream、正文按data:分块到达。三者缺一,就说明链路某一段有问题。
把这两步都跑通之后,你的 MCP 服务就具备了"被大模型调用"和"调用大模型"的双向能力。实际项目里,你可以把get_forecast换成自己的业务函数,比如查数据库、读设备状态,MCP 会自动生成工具描述,大模型就能理解怎么调用。
5. 本篇常见错排查:401、local proxy failed、reading choices 与 OAuth
这一节按真实报错来排,每个都给出定位思路。
401 Unauthorized:最常见的原因是 Key 没读到或者格式不对。先确认环境变量是否在当前 shell 生效,echo $TAOTOKEN_API_KEY看有没有值。如果值存在但仍然 401,检查请求头是不是Authorization: Bearer sk-xxx,Bearer 和 Key 之间有一个空格,少空格会直接失败。还有一种情况是 Key 被复制时带了换行或空格,用tr -d ' \n'清理一下。
local proxy failed:这个报错通常出现在客户端侧,说明客户端尝试通过本地代理连接 MCP 服务但失败了。检查你的 MCP 服务是否真的在监听,curl http://127.0.0.1:8001/mcp看有没有响应。如果服务正常,检查客户端配置里的 URL 是不是写成了localhost而服务只绑定了127.0.0.1,两者在某些系统上不等价,统一用127.0.0.1更稳。另外确认端口没有被其他进程占用,lsof -i :8001可以查。
reading choices 相关报错:这类错误一般出现在解析大模型响应时,提示读取choices字段失败。根因通常是返回体不是预期的 JSON 结构,可能是 Base URL 配错导致打到了非兼容接口,或者 Model ID 不存在返回了错误对象。先用第 4 节的 curl 命令单独验证通道,确认返回体里有choices数组再回到 MCP 代码。如果 curl 正常但代码报错,检查你的 HTTP 客户端有没有正确设置stream=True。
OAuth 相关报错:部分 MCP 客户端在连接时会尝试 OAuth 流程,如果你的服务没有实现鉴权端点,就会卡在授权环节。对于本地开发,最简单的做法是在客户端配置里关闭 OAuth 或者选择"无鉴权"模式。如果你确实需要鉴权,再单独实现,不要和通道验证混在一起做,否则排障会很痛苦。
排障的通用原则是分层验证:先验证大模型通道(curl),再验证 MCP 服务(inspector),最后验证客户端到 MCP 的连接。每一层单独确认,不要跳步。接入相关的详细文档可以参考 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有各客户端的配置示例。
6. 语义一致 CTA:把这条链路用到你的真实项目里
链路跑通只是起点。接下来你可以做几件很实际的事:把get_forecast替换成你自己的业务函数,比如查询订单状态、读取传感器数据、触发设备动作;把config.toml里的model_id换成更适合你场景的模型;如果你的调用量上来了,考虑用 Coding Plan 来降低单位成本。
对于需要长期跑 Agent 的场景,建议把 MCP 服务和模型通道分开部署,MCP 服务负责工具暴露,模型通道负责推理调用,两者通过环境变量解耦。这样换模型时不用重启 MCP 服务,改一个环境变量就行。
如果你在接入过程中遇到具体的报错,优先去 API Keys 页面确认 Key 状态,再去接入文档对照配置。模型对话页面可以用来快速验证某个模型是否可用,避免在代码里反复试错。把这三件套(Base URL、Key、Model ID)管理好,你的大模型通信链路就能稳定运行,后续扩展工具也只是加一个函数的事。