企业官网建设流程全解析

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 pandas

FastMCP 对 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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.txt

server.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( f"https://api.example.com/nav", params={"code": fund_code, "start": start_date, "end": end_date}, timeout=10 ) records = resp.json()["data"] conn = sqlite3.connect("cache.db") pd.DataFrame(records).to_sql("nav", conn, if_exists="append", index=False) conn.close() return {"source": "remote", "data": records} if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=8001)

关键点在最后一行。transport="streamable-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}, timeout=10 ) items = resp.json()["items"] return [ {"title": i["title"], "summary": i["summary"], "date": i["date"]} for i in items ] if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=8002)

两个 Server 分开跑的好处是：新闻接口不稳定时不会拖垮净值查询；而且可以独立扩缩容，新闻检索加缓存层、净值查询加数据库索引，互不影响。

3.3 Streamable HTTP 接入参数对照

客户端连接时需要的参数不多，但容易填错。我整理了一张对照表：

客户端配置示例（以 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

启动两个 Server：

python 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": f"Bearer {TAOTOKEN_KEY}"}, json={ "model": "gemini-2.5-flash", "messages": [{"role": "user", "content": "分析 110011 这只基金近一个月的涨跌原因"}], "tools": tools }, timeout=30 ) 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 Type

Streamable HTTP 要求请求头带Accept: application/json, text/event-stream。有些 HTTP 客户端默认只发application/json，服务端协商失败。在客户端配置里显式加上这个头。

报错三：模型不调用工具，直接编答案

检查 tools 的 schema 是否符合 JSON Schema 规范。常见问题是required字段拼写错误，或者参数类型写成 Python 的str而不是 JSON 的string。另外，Gemini 对工具描述的语义敏感，description 写清楚「什么时候该用这个工具」，比写「这个工具做什么」更有效。

报错四：TaoToken 返回 401

Key 没读到或者过期。先确认环境变量注入成功：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_exists="append"会自动建表。

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。这个在本地调试时不容易发现，部署到云上跑一段时间才会暴露。