1. 从零跑通大模型通信链路FastAPI SSE MCP 服务到底解决什么问题如果你正在做 AI Agent 或者智能硬件后端大概率会遇到一个很具体的场景你有一堆现成的 Python 业务接口查天气、查订单、读传感器数据想让大模型直接调用它们但每次对接都要手写一套工具描述、参数校验、流式返回改一个字段就要动三四个文件。MCPModel 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 了解整体能力然后进入控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建 Key再到 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 管理你的密钥。如果你只是想先验证模型能不能通可以直接用模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 发一条消息试试确认通道没问题再写代码。环境变量建议这样设置Linux/macOS 用 exportWindows 用 setexport TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://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_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它在高频调用下更划算如果只是偶尔验证按量调用即可。把这三件套准备好之后你的 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(titleMCP Gateway) mcp_server add_mcp_server( app, mount_pathcfg[mcp][mount_path], namecfg[mcp][name], describe_all_responsescfg[mcp][describe_all_responses], describe_full_response_schemacfg[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/geojson} async with httpx.AsyncClient() as client: try: resp await client.get(url, headersheaders, timeoutcfg[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/apiKey 填环境变量引用Model ID 填你在控制台选的模型。三件套缺一不可尤其是 Model ID漏填会直接报模型不存在。4. 验证请求一次真实的 SSE 流式调用与成功结果服务起来之后必须做一次真实的 SSE 请求验证否则你不知道链路到底通没通。这里分两步先验证 MCP 服务本身再验证大模型通道。第一步用 MCP Inspector 连接 SSE 端点。启动 inspectorCLIENT_PORT8081 SERVER_PORT8082 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-xxxBearer 和 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 客户端有没有正确设置streamTrue。OAuth 相关报错部分 MCP 客户端在连接时会尝试 OAuth 流程如果你的服务没有实现鉴权端点就会卡在授权环节。对于本地开发最简单的做法是在客户端配置里关闭 OAuth 或者选择无鉴权模式。如果你确实需要鉴权再单独实现不要和通道验证混在一起做否则排障会很痛苦。排障的通用原则是分层验证先验证大模型通道curl再验证 MCP 服务inspector最后验证客户端到 MCP 的连接。每一层单独确认不要跳步。接入相关的详细文档可以参考 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各客户端的配置示例。6. 语义一致 CTA把这条链路用到你的真实项目里链路跑通只是起点。接下来你可以做几件很实际的事把get_forecast替换成你自己的业务函数比如查询订单状态、读取传感器数据、触发设备动作把config.toml里的model_id换成更适合你场景的模型如果你的调用量上来了考虑用 Coding Plan 来降低单位成本。对于需要长期跑 Agent 的场景建议把 MCP 服务和模型通道分开部署MCP 服务负责工具暴露模型通道负责推理调用两者通过环境变量解耦。这样换模型时不用重启 MCP 服务改一个环境变量就行。如果你在接入过程中遇到具体的报错优先去 API Keys 页面确认 Key 状态再去接入文档对照配置。模型对话页面可以用来快速验证某个模型是否可用避免在代码里反复试错。把这三件套Base URL、Key、Model ID管理好你的大模型通信链路就能稳定运行后续扩展工具也只是加一个函数的事。