1. 从一次真实的踩坑说起为什么我要手搓两个 MCP Server去年底我给自己写了个基金涨跌分析的小工具前端用 Next.js后端 Flask数据靠爬。跑通之后发现一个尴尬的问题大模型拿不到实时数据。我把净值曲线截图丢给模型它能说出一堆正确的废话但问它「这只基金上周三为什么跌了 2.3%」它就开始编。后来我把数据接口直接塞进 prompt问题更大了。基金净值是结构化的时间序列新闻是非结构化的文本两类数据混在一起token 消耗飞快模型还经常把 A 基金的涨跌安到 B 基金头上。这就是典型的上下文污染。MCP Server 解决的正是这件事。它把「取数据」和「用数据」拆开模型通过标准协议按需调用工具拿到的是干净的、结构化的、可追溯的上下文。我最终拆成两个 Server一个管基金净值一个管相关新闻。传输方式选 Streamable HTTP因为要部署到云上给多个客户端用。模型侧用 Gemini通过 TaoToken 统一走一个 Key 和 API 通道省得在 Google 开发者后台反复配项目。这篇文章会带你从零搭出这两个 Server 的骨架配好 Streamable HTTP 的接入参数跑一次可复现的调用验证最后把模型侧接上。适合有 Python 基础、想给自己的工具接大模型但不想被 RAG 检索坑过的独立开发者。2. 前置准备FastMCP 环境与 TaoToken 通道2.1 为什么选 FastMCP 而不是裸写协议MCP 协议本身不复杂但手写 JSON-RPC 的消息路由、能力协商、会话管理很烦。FastMCP 把这些都封好了你只需要用装饰器标记工具函数它自动生成 schema、处理参数校验、管理 Streamable HTTP 的会话生命周期。我试过裸写一遍光 session 复用就调了一下午换 FastMCP 之后二十分钟跑通。安装很简单pip install fastmcp httpx pandasFastMCP 对 Python 版本要求 3.10我本地是 3.11没遇到兼容问题。如果你用虚拟环境记得在启动脚本里显式指定解释器路径后面配 Streamable HTTP 的时候会用到。2.2 TaoToken 在链路里的位置模型侧我选 Gemini-2.5-flash原因是它对 function calling 的支持比较稳而且响应快。但直接对接 Google 的接口有个麻烦每个环境都要单独配 Key本地调试、服务器部署、CI 测试三套凭证管理成本高。TaoToken 在这里的角色是统一通道。你拿一个 Key通过它的 API 地址调用模型底层走哪个厂商对上层透明。官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 Key。API 地址是 https://taotoken.net/api 注意这个不带 UTM 参数直接填进代码里。具体操作登录后进控制台左侧找 API Keys新建一个复制出来。这个 Key 后面会同时用在 Gemini 调用和 MCP Server 的鉴权上。如果你打算长期跑编码类任务可以看看 Coding Plan它针对高频调用做了额度优化只是验证模型连通性的话用模型对话页面手动测一次就行。注意Key 不要硬编码进代码提交到 Git。我用的是环境变量加 .env 文件.env 写进 .gitignore。3. 可复制配置两个 MCP Server 的骨架与 Streamable HTTP 参数3.1 Fund Server净值数据的结构化封装先建目录结构fund-mcp/ server.py cache.db requirements.txtserver.py 的核心逻辑是接收基金代码和时间段先查本地 SQLite 缓存命中就直接返回没命中就调外部数据源补齐写入缓存后再返回。这样重复分析同一只基金时响应从秒级降到毫秒级。from fastmcp import FastMCP import sqlite3, httpx, pandas as pd from datetime import datetime mcp FastMCP(FundServer) def get_cache(fund_code, start, end): conn sqlite3.connect(cache.db) df pd.read_sql( SELECT * FROM nav WHERE code? AND date BETWEEN ? AND ?, conn, params(fund_code, start, end) ) conn.close() return df mcp.tool() def get_fund_nav(fund_code: str, start_date: str, end_date: str) - dict: 获取指定基金在时间段内的净值与涨跌幅 cached get_cache(fund_code, start_date, end_date) if len(cached) 0: return {source: cache, data: cached.to_dict(records)} # 缓存未命中走外部接口补齐 resp httpx.get( fhttps://api.example.com/nav, params{code: fund_code, start: start_date, end: end_date}, timeout10 ) records resp.json()[data] conn sqlite3.connect(cache.db) pd.DataFrame(records).to_sql(nav, conn, if_existsappend, indexFalse) conn.close() return {source: remote, data: records} if __name__ __main__: mcp.run(transportstreamable-http, host0.0.0.0, port8001)关键点在最后一行。transportstreamable-http让 FastMCP 用 Streamable HTTP 暴露服务默认端点是/mcp。host 设 0.0.0.0 是为了容器内可访问本地调试可以改 127.0.0.1。3.2 News Server新闻检索与相关性过滤第二个 Server 独立进程端口错开from fastmcp import FastMCP import httpx mcp FastMCP(NewsServer) mcp.tool() def search_fund_news(fund_code: str, keywords: str, limit: int 5) - list: 按基金代码和关键词检索相关新闻返回标题、摘要、时间 resp httpx.get( https://api.example.com/news, params{code: fund_code, q: keywords, size: limit}, timeout10 ) items resp.json()[items] return [ {title: i[title], summary: i[summary], date: i[date]} for i in items ] if __name__ __main__: mcp.run(transportstreamable-http, host0.0.0.0, port8002)两个 Server 分开跑的好处是新闻接口不稳定时不会拖垮净值查询而且可以独立扩缩容新闻检索加缓存层、净值查询加数据库索引互不影响。3.3 Streamable HTTP 接入参数对照客户端连接时需要的参数不多但容易填错。我整理了一张对照表参数Fund ServerNews Server说明传输方式streamable-httpstreamable-http必须一致端点路径/mcp/mcpFastMCP 默认端口80018002避免冲突超时15s20s新闻接口较慢重试2 次3 次网络抖动容忍客户端配置示例以 Python 客户端为例from fastmcp import Client fund_client Client(http://127.0.0.1:8001/mcp) news_client Client(http://127.0.0.1:8002/mcp)如果你部署在云服务器把 127.0.0.1 换成公网 IP 或域名。生产环境建议加一层反向代理做 TLS 终止Streamable HTTP 本身支持 HTTP/2代理配置里记得开启。4. 验证请求一次可复现的调用与成功结果4.1 先单独验证 MCP Server启动两个 Serverpython fund-mcp/server.py python news-mcp/server.py 用 curl 发一个初始化请求确认 Streamable HTTP 端点活着curl -X POST http://127.0.0.1:8001/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:test,version:1.0}}}正常返回里会有serverInfo和capabilities字段。如果返回 404检查路径是不是/mcp如果连接被拒检查端口和防火墙。4.2 再验证模型侧调用模型侧通过 TaoToken 调 Gemini把两个 MCP Server 的工具描述作为 function 传入。核心代码import httpx, os TAOTOKEN_KEY os.getenv(TAOTOKEN_KEY) API_URL https://taotoken.net/api/v1/chat/completions tools [ { type: function, function: { name: get_fund_nav, description: 获取基金净值与涨跌幅, parameters: { type: object, properties: { fund_code: {type: string}, start_date: {type: string}, end_date: {type: string} }, required: [fund_code, start_date, end_date] } } }, { type: function, function: { name: search_fund_news, description: 检索基金相关新闻, parameters: { type: object, properties: { fund_code: {type: string}, keywords: {type: string}, limit: {type: integer} }, required: [fund_code, keywords] } } } ] resp httpx.post( API_URL, headers{Authorization: fBearer {TAOTOKEN_KEY}}, json{ model: gemini-2.5-flash, messages: [{role: user, content: 分析 110011 这只基金近一个月的涨跌原因}], tools: tools }, timeout30 ) print(resp.json())成功的话返回里会看到tool_calls字段模型决定先调get_fund_nav拿净值再调search_fund_news拿新闻。你把这几个调用结果回填给模型它就能生成一段有数据支撑的涨跌解读而不是空泛的套话。实测下来从发起到拿到完整分析端到端在 3 到 5 秒主要耗时在新闻接口。净值走缓存的话基本无感。5. 本篇常见错排查报错一Connection refused或ClientConnectorError九成是 Server 没起来或者端口填错。先lsof -i:8001确认进程在监听。如果 Server 起来了还是连不上检查是不是绑到了 127.0.0.1 而客户端在容器里访问。把 host 改成 0.0.0.0 重启。报错二406 Not Acceptable或Unsupported Media TypeStreamable HTTP 要求请求头带Accept: application/json, text/event-stream。有些 HTTP 客户端默认只发application/json服务端协商失败。在客户端配置里显式加上这个头。报错三模型不调用工具直接编答案检查 tools 的 schema 是否符合 JSON Schema 规范。常见问题是required字段拼写错误或者参数类型写成 Python 的str而不是 JSON 的string。另外Gemini 对工具描述的语义敏感description 写清楚「什么时候该用这个工具」比写「这个工具做什么」更有效。报错四TaoToken 返回 401Key 没读到或者过期。先确认环境变量注入成功echo $TAOTOKEN_KEY。如果 Key 是对的还报 401检查请求头格式必须是Bearer加空格再加 Key。另外注意 API 地址结尾不要多加斜杠/api/v1/chat/completions是完整路径。报错五缓存表不存在第一次跑 Fund Server 时cache.db里还没有nav表pd.read_sql会抛异常。在get_cache里加个 try-except表不存在时返回空 DataFrame让流程走远程补齐分支写入时if_existsappend会自动建表。6. 模型侧接入与后续扩展模型侧接入的核心就三件事拿 Key、配地址、传工具。Key 在 TaoToken 控制台生成地址用 https://taotoken.net/api 工具描述从 FastMCP 的mcp.get_tools()里导出转成 OpenAI 兼容的 function 格式。这样你的 Flask 后端只需要维护一份工具列表两个 MCP Server 增减工具时模型侧自动同步。如果你打算把这个工具做成长期跑的服务建议把 Key 管理、调用日志、额度监控都收拢到 TaoToken 的控制台里看。我自己的做法是本地开发用模型对话页面手动验证 prompt 效果确认没问题再写进代码走 API。编码类的高频调用场景可以看 Coding Plan普通分析类调用按量走 API 就行。接入文档在 https://taotoken.net/doc 里面有各语言 SDK 的示例。API Keys 管理页在 https://taotoken.net/api-keys 建议给不同环境建不同的 Key方便排查问题时定位来源。最后说一个我踩过的坑Streamable HTTP 的会话在长时间空闲后会被服务端回收客户端需要实现重连逻辑。FastMCP 的 Client 默认带重试但如果你自己裸写 HTTP 请求记得在收到 session 失效的错误码时重新 initialize。这个在本地调试时不容易发现部署到云上跑一段时间才会暴露。