1. 水质数据查询为什么需要一个 MCP Server水质在线数据查询这件事麻烦的地方不在数据本身而在于数据散落在不同地方有的在环保部门的公开 API 里有的在自建 MySQL 里还有的躺在运维同事的 Excel 里。每次想让 AI 帮忙分析一下某站点最近氨氮有没有超标都得手动导出、粘贴、再问模型来回折腾。MCP Server 就是来解决这个断层的。MCPModel Context Protocol可以理解成 AI 世界的 USB-C 接口模型本身只会说话但通过 MCP 协议它能以标准方式调用你定义好的工具去拿实时数据、执行查询。你把手头的水质查询逻辑封装成一个 MCP 工具模型就能在对话里直接调用它拿到结构化结果再组织成人话回答你。这篇要做的是手搓一个水质在线数据查询的 MCP Server然后通过 TaoToken 统一 Key/API 通道把它接进 AI 工具链。TaoToken 在这里扮演的是统一入口的角色——你不用为每个工具单独配一堆 Key而是走同一个 API 通道模型对话、编码 Agent、MCP 工具调用都能复用。适合谁看手上有水质/环境监测数据、想让 AI 直接查、又不想每次手动喂数据的开发和运维同学。整条链路的目标很明确从写 MCP Server到配 config.toml / settings.json再到 CC Switch 和 Cline 里挂上最后发一次真实查询请求看到数据返回。跑通一次后面换数据源只是改函数的事。2. TaoToken 前置准备Key 与通道在写代码之前先把 TaoToken 这边的入口准备好。MCP Server 本身是本地进程但它要调用模型能力时走的是 TaoToken 的统一 API 通道所以你需要一个可用的 Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册登录后进控制台。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在里面找到 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。新建一个 Key复制出来存好后面配置里要用。API 的基础地址是 https://taotoken.net/api 注意这个地址不带任何查询参数配置时直接填这个就行。如果你用的是兼容 OpenAI 风格的客户端通常还需要在末尾拼 /v1具体以你所用工具的文档为准。注意Key 只显示一次丢了只能重建。建议按用途分开建 Key比如一个给 MCP 工具用一个给编码 Agent 用方便排查和限额。模型对话想先验证通道通不通可以直接用模型对话页面https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 发一句话看有没有正常回复。这一步能快速确认 Key 和通道没问题再去折腾 MCP 配置省得两头排查。如果你后面要长期跑编码类 Agent比如让它自动改 MCP Server 代码可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入细节和参数说明都在文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 手搓水质查询 MCP Server3.1 项目初始化与依赖用 uv 管理项目比 pip 干净。先建目录再初始化uv init water_mcp cd water_mcp uv add mcp[cli] httpx pandasmcp[cli]提供 FastMCP 和调试命令httpx 用来请求远程水质 APIpandas 处理表格数据。如果你数据在 MySQL再加pymysql。3.2 核心代码把查询封装成工具新建server.py核心思路是把「取数据」和「格式化输出」分开。取数据函数负责对接真实数据源工具函数负责接收模型传来的参数、调用取数、返回可读文本。import pandas as pd from datetime import datetime from mcp.server.fastmcp import FastMCP mcp FastMCP(water_quality) def fetch_station_data(station: str, start: datetime, end: datetime) - pd.DataFrame: 从真实数据源取数这里用模拟数据占位。 实际替换为 httpx 请求 API 或 pymysql 查询即可。 data { 监测时间: [datetime(2025, 5, 1, 10, 0), datetime(2025, 5, 1, 11, 0)], 设备名称: [station, station], 氨氮: [0.5, 0.6], 总磷: [0.2, 0.25], 高锰酸盐指数: [2.1, 2.3], } return pd.DataFrame(data) mcp.tool() async def get_water_quality( start_time: str, end_time: str, station_name: str, factors: str None, ) - str: 查询指定站点和时间范围内的水质监测数据。 start_time/end_time 格式: YYYY-MM-DD HH:MM:SS factors 为逗号分隔的因子名如 氨氮,总磷可选。 try: start datetime.strptime(start_time, %Y-%m-%d %H:%M:%S) end datetime.strptime(end_time, %Y-%m-%d %H:%M:%S) df fetch_station_data(station_name, start, end) if df is None or df.empty: return 未找到符合条件的监测数据。 df[监测时间] pd.to_datetime(df[监测时间]) if factors: wanted [f.strip() for f in factors.split(,)] valid [f for f in wanted if f in df.columns] if not valid: return 指定的因子在监测数据中不存在。 df df[[监测时间, 设备名称] valid] return df.to_string(indexFalse) except ValueError as e: return f参数错误: {e} except Exception as e: return f发生错误: {e} if __name__ __main__: mcp.run(transportstdio)几个关键点。mcp.tool()装饰器把函数注册成模型可调用的工具函数签名里的类型标注和 docstring 就是模型理解「怎么调、传什么」的依据所以 docstring 要写清楚参数格式。transportstdio表示用标准输入输出通信这是本地 MCP 最常用的方式AI 工具启动这个进程后通过管道对话。fetch_station_data是唯一需要你改的地方。换成 httpx 请求import httpx def fetch_station_data(station, start, end): resp httpx.get( https://your-water-api.example.com/query, params{station: station, start: start.isoformat(), end: end.isoformat()}, timeout30.0, ) resp.raise_for_status() return pd.DataFrame(resp.json()[data])换成 MySQL 就把pd.DataFrame换成pd.read_sql的结果。工具层完全不用动。3.3 本地自测跑之前先用 MCP 自带的调试器确认工具能被识别uv run mcp dev server.py它会起一个本地调试界面能看到get_water_quality出现在工具列表里手动填参数调用一次确认返回的是格式化表格而不是报错。这一步过了再往 AI 工具里挂。4. 可复制配置config.toml 与 settings.json4.1 config.toml 骨架有些工具用 TOML 配置 MCP Server骨架长这样[mcp_servers.water_quality] command uv args [--directory, /Users/you/water_mcp, run, server.py] env { TAOTOKEN_API_KEY sk-你的Key, TAOTOKEN_BASE_URL https://taotoken.net/api }--directory指向项目绝对路径run server.py让 uv 在项目环境里执行。env 里把 TaoToken 的 Key 和 base url 传进去Server 内部如果要调模型或转发请求就能直接读环境变量不用硬编码。4.2 settings.json 骨架Cline、Claude Code 这类工具用 JSON{ mcpServers: { water_quality: { command: uv, args: [--directory, /Users/you/water_mcp, run, server.py], env: { TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }路径一定要用绝对路径相对路径在不同工具的工作目录下会找不到文件这是最常见的挂载失败原因。4.3 CC Switch 配置片段CC Switch 用来在多个模型通道间切换。把 TaoToken 作为一个 provider 加进去指向 API 地址{ providers: [ { name: taotoken, base_url: https://taotoken.net/api, api_key: sk-你的Key, models: [claude-sonnet-4-20250514] } ] }切到 taotoken 这个 provider 后MCP 工具调用和模型对话都走同一条通道Key 管理集中在一处。Claude Code 相关的接入方式在文档里有专门说明https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。4.4 Cline 配置片段Cline 的 MCP 配置在设置面板里格式和 settings.json 一致。把上面那段mcpServers贴进去保存后 Cline 会尝试启动进程。启动成功的话工具列表里会出现water_quality展开能看到get_water_quality这个工具及其参数说明。5. 验证请求发一次真实查询配置挂好后别急着问复杂问题先用一句明确的指令验证链路。在 Cline 或 Claude Code 的对话里输入查询站点 A 在 2025-05-01 10:00:00 到 2025-05-01 12:00:00 的氨氮和总磷数据模型会做几件事识别出要调用get_water_quality从自然语言里抽出station_nameA、start_time、end_time、factors氨氮,总磷发起工具调用拿到返回的表格文本再组织成回答。成功的标志是你能看到工具调用记录多数工具会显示「正在调用 get_water_quality」然后返回类似监测时间 设备名称 氨氮 总磷 2025-05-01 10:00:00 站点A 0.5 0.20 2025-05-01 11:00:00 站点A 0.6 0.25如果模型没调工具、直接编了一段数据说明工具没挂上或描述没被识别回到第 4 步检查配置。如果调用了但报错看第 6 节的排查。想单独验证模型通道是否正常可以先用模型对话页面发一条普通消息https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。通道正常但工具不触发问题就锁定在 MCP 配置侧。6. 本篇常见错排查工具列表里看不到 water_quality。九成是路径问题。--directory必须是绝对路径且该目录下有server.py。在终端手动跑一遍uv --directory /你的路径 run server.py如果报错工具侧也一样起不来。进程启动就退出。多半是依赖没装全。回到项目目录uv sync或uv add补齐确认uv run server.py能挂起等待输入stdio 模式下它会阻塞等消息这是正常的。调用返回「参数错误」。时间格式没对上。docstring 里写的是YYYY-MM-DD HH:MM:SS模型有时会传2025/05/01在工具里加一层兼容解析或者把 docstring 写得更死板一点。返回「指定的因子在监测数据中不存在」。因子名和 DataFrame 列名不一致比如数据里叫「氨氮(mg/L)」而你传「氨氮」。统一列名或在工具里做模糊匹配。Key 报 401。检查 env 里的TAOTOKEN_API_KEY有没有多余空格base url 是不是https://taotoken.net/api不带 UTM 参数。Key 失效就去 API Keys 页面重建https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。模型不调工具、自己编数据。工具描述太模糊。把 docstring 写具体明确「必须调用此工具获取真实数据不要自行推测」多数模型会照做。7. 把链路固定下来跑通一次之后建议把fetch_station_data里的模拟数据换成真实数据源然后把这个 MCP Server 的配置固化到你的工具链里。长期跑编码类 Agent 自动维护这个 Server 的话Coding Plan 会比按次调用更省心https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。我自己的习惯是数据源查询逻辑单独放一个模块MCP 工具层只做参数校验和格式化这样换数据源时工具层零改动。另外把 Key 放环境变量而不是写死在配置里换机器时只改一处。水质数据这类带时间戳的查询记得在工具里对时间范围做个上限保护避免模型传个跨十年的区间把数据源拖垮。
建堆推导)