1. 数据分析智能体接 MCP 时,我踩到的第一个坑
数据分析智能体要真正跑起来,绕不开一个现实问题:模型本身不会查库、不会算指标,它只会“说”。所以你需要给它一双手,让它能调用你写好的函数去取数、清洗、统计。MCP(Model Context Protocol)就是干这个的——它把工具能力标准化,让大模型按统一协议发现并调用你的函数。而 fastmcp 是把这件事做得最省心的 Python 框架之一,几行装饰器就能把一个异步函数暴露成工具。
但工具能跑通,不代表链路能跑通。真正卡人的地方在于:模型侧要有一个稳定的 Key 通道,工具侧要有一个能被 Open-WebUI 识别的代理层,中间还要处理 SSE 长连接、参数校验、跨天时间范围这些细节。我这次的目标很明确——用 fastmcp 写一个“使用率查询”工具,通过 MCPO 代理成 OpenAPI,再挂到 Open-WebUI 的工具服务器里,同时把模型请求统一走 TaoToken 的 Key 通道。适合谁看?正在做数据分析 Agent、准备把 MCP 工具接进 Open-WebUI、或者被 SSE 和 config.toml 折腾过的朋友。
下面按我实际踩坑的顺序来:先讲清楚工具函数为什么要写得“啰嗦”,再讲 TaoToken 的 Key 怎么配,然后是 MCPO 代理和 Open-WebUI 的完整配置,最后是连通性验证和几个把我卡了半天的报错。
2. 前置准备:TaoToken 统一 Key 通道与 MCP 工具定位
在写工具之前,先把模型侧的通道定下来。数据分析智能体的特点是“多轮调用 + 工具回传”,如果每次请求都散落在不同 Key 上,排查问题会非常痛苦。我这次把所有模型请求收敛到 TaoToken 的统一 Key 通道,好处是:一个 Key 管所有模型调用,工具调用和普通对话走同一出口,日志和额度也能对得上。
TaoToken 的接入地址是https://taotoken.net/api,控制台在https://taotoken.net/console,API Keys 管理在https://taotoken.net/api-keys。如果你用的是 Claude Code 这类编码场景,可以看https://taotoken.net/claude-code-anthropic;想先验证模型通不通,直接开https://taotoken.net/model-chat对话页试一句就行;长期跑 Agent 或编码任务,https://taotoken.net/coding-plan更适合。
这里要强调一个定位:TaoToken 是模型调用的统一入口,不是替代你本地 MCP 服务的运行时。MCP 工具仍然跑在你自己的机器上(比如localhost:8001),TaoToken 负责的是“模型怎么被调用”。两者是上下游关系,别混在一起配。
注意:MCP 服务不要直连生产数据库。我这次的工具函数内部走的是只读查询 + 聚合计算,参数里也做了行政区归属校验,避免模型乱传参数打到核心库。
3. 可复制配置:fastmcp 工具 + MCPO 代理 + Open-WebUI
3.1 用 fastmcp 写一个“参数啰嗦”的工具
我试过把参数描述写得很简略,结果模型经常把time_range传成null,有时候传start_time=null & end_time=null,有时候干脆不传。后来干脆统一标准:让模型必须显式提供完整的时间范围,全天就填00:00:00-23:59:59。参数描述越详尽,模型理解歧义越小。
from typing import Annotated, List from pydantic import Field from fastmcp import FastMCP mcp = FastMCP("data-analyst-mcp") class Property(dict): """地理位置/组织机构/所有权组合筛选条件""" class DateRange(dict): """日期范围,必须同时提供 start_date 和 end_date""" class TimeRange(dict): """时间范围,必须同时提供 start_time 和 end_time,支持跨天""" @mcp.tool() async def get_usage_rate( property_list: Annotated[List[Property], Field( description="每个成员通过地理位置(province/city/district)、组织机构(organization)、所有权(ownership_type)筛选目标。成员数大于1时视为批量查询,取并集", examples=[[{"province": "XX省", "ownership_type": "自营"}, {"district": "XX区", "organization": "XX公司", "ownership_type": "合营"}]] )], date_range: Annotated[DateRange, Field( description="日期范围,必须同时提供 start_date 和 end_date", examples=[{"start_date": "2023-01-01", "end_date": "2023-01-31"}] )], time_range: Annotated[TimeRange, Field( description="时间范围,必须同时提供 start_time 和 end_time。全天填 00:00:00-23:59:59,支持跨天如 23:30:00-09:00:00", examples=[{"start_time": "08:00:00", "end_time": "18:00:00"}, {"start_time": "23:30:00", "end_time": "09:00:00"}] )], ) -> list[dict]: """ 查询满足指定属性的所有目标在指定时间段的【使用率】。 规则:至少提供一种属性;行政区父子级冲突返回错误;日期和时间范围必须完整。 返回:[[组织机构, 所有权, 省, 市, 区县, 编码, 使用率], ...] """ # 实际实现:参数校验 -> 只读查询 -> 聚合计算 return []关键点在于Annotated + Field的description和examples。模型是靠这些文字来决定怎么填参数的,写得越像“给新人看的接口文档”,调用成功率越高。time_range我一开始允许为None表示全天,结果模型行为不稳定,后来强制要求显式传值,问题就消失了。
3.2 用 SSE 模式启动 MCP 服务
fastmcp 支持 stdio 和 SSE 两种传输。Open-WebUI 这边需要 HTTP 可达,所以用 SSE:
# 启动 SSE 服务,监听 8001 python -m fastmcp run server.py --transport sse --port 8001启动后你会看到类似Uvicorn running on http://0.0.0.0:8001的日志,SSE 端点是http://localhost:8001/sse。
3.3 MCPO 代理:把 MCP 转成 OpenAPI
Open-WebUI 不能直接吃 MCP 协议,需要 MCPO 做一层代理,把 MCP 工具转成 OpenAPI 接口。先配config.json:
{ "mcpServers": { "data-analyst-mcp": { "url": "http://localhost:8001/sse" } } }然后设置国内镜像源并启动 MCPO:
export UV_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple uvx mcpo --port 8000 --config ./config.json成功的话,访问http://localhost:8000/docs能看到自动生成的 OpenAPI 文档,里面会有get_usage_rate这个接口。
3.4 Open-WebUI 工具服务器配置
在 Open-WebUI 里进入“工具服务器”,新增一个连接,地址填http://localhost:8000,保存后启用。此时 Open-WebUI 会拉取 MCPO 暴露的 OpenAPI schema,把get_usage_rate注册成可调用工具。
如果你用的是 Cline 或 CC Switch 这类客户端,配置片段类似:
{ "mcpServers": { "data-analyst-mcp": { "url": "http://localhost:8001/sse" } } }Cline 的settings.json里则是把工具服务器地址指向 MCPO 的http://localhost:8000,模型侧 Base URL 填https://taotoken.net/api,Key 用你在https://taotoken.net/api-keys生成的那把。
4. 验证请求:从连通性到工具真正被调用
配置完别急着上复杂提示词,先做三层验证。
第一层,验证 MCP 服务本身活着:
curl -N http://localhost:8001/sse能看到 SSE 事件流持续输出,说明 fastmcp 正常。
第二层,验证 MCPO 代理通了:
curl http://localhost:8000/docs返回 OpenAPI 页面,且/get_usage_rate出现在接口列表里。
第三层,验证模型能调用工具。在 Open-WebUI 里输入提示词:
帮我查一下 XX 省自营设备 2023 年 1 月全天的使用率
如果工具被顺利调用,你会在界面上看到工具调用卡片,参数里time_range是{"start_time": "00:00:00", "end_time": "23:59:59"},返回结果是数组列表。这一步跑通,说明“模型 → TaoToken → 工具 → MCPO → fastmcp”整条链路是通的。
5. 本篇常见错排查
报错一:MCPO 启动后/docs打不开。多半是config.json里的 URL 写成了http://localhost:8001而不是http://localhost:8001/sse。SSE 端点必须带/sse后缀。
报错二:Open-WebUI 里工具列表为空。检查 MCPO 是否真的拉到了工具。可以先curl http://localhost:8000/openapi.json,看 schema 里有没有你的函数。没有的话,回到 fastmcp 确认@mcp.tool()装饰器生效、函数是async的。
报错三:模型调用工具时报参数校验失败。典型是time_range传了null。解决办法就是前面说的——在Field描述里明确写“必须同时提供 start_time 和 end_time”,并给全天和跨天两个 examples。模型对 examples 的敏感度比纯文字描述高。
报错四:跨天时间范围算错。比如23:30:00-09:00:00,如果内部逻辑没处理跨天,会把结束时间当成当天早上。工具函数里要显式判断end_time < start_time时加一天。
报错五:模型侧 401 或连不上。检查 Base URL 是否为https://taotoken.net/api,Key 是否从https://taotoken.net/api-keys正确复制。如果只是想先验证模型通不通,去https://taotoken.net/model-chat发一句话最快。
6. 后续怎么走:把工具做厚,把通道做稳
这次只放了一个get_usage_rate,但结构已经搭好了。下一步我会把 ETL、清洗、统计拆成更多细粒度工具,让模型自己组合调用,而不是一个函数包办所有事——函数越“重”,模型越容易在参数上翻车。工具描述继续按“给新人写接口文档”的标准来,examples 至少给两个。
模型通道这边,长期跑 Agent 建议直接用https://taotoken.net/coding-plan,省得每次手动管额度;接入文档在https://taotoken.net/doc,配置细节都在里面。等整套在内网跑稳,再把 MCPO 和 fastmcp 一起挪过去,Key 通道保持不变,迁移成本几乎为零。