1. 为什么我要自己写一个 ChatBI MCP Server先说清楚这东西是什么。ChatBI MCP Server 是一个把「自然语言查数」能力封装成标准工具协议的服务端程序它让 Claude、Cursor、Cherry Studio 这类支持 MCP 的客户端能够通过一次对话直接读取你的 Excel/CSV 文件、生成分析代码、跑出统计结果。适合谁适合手头有一堆业务表格、又不想每次都写 pandas 脚本的开发者也适合正在学 Agent 工具调用、想找一个能完整跑通的实战项目的人。我试过直接让大模型读 CSV 然后口算数据量一过千行就开始胡说列名稍微复杂点就张冠李戴。原因很简单模型没有真正执行代码它在做概率补全。ChatBI 的思路是把「生成代码」和「执行代码」分开——模型只负责根据数据描述写 pandas 代码真正的计算交给本地 Python 进程。这样准确率是数量级的提升。MCPModel Context Protocol在这里扮演的角色是「工具插座」。你不需要为每个客户端写一套适配层只要按协议暴露工具任何支持 MCP 的客户端都能即插即用。我们要实现两个核心工具get_preview_data负责数据探查把列名、类型、前几行样本以 Markdown 形式返回给模型analyze_data负责接收自然语言问题内部调用大模型生成代码并执行返回结果表格。整条链路里大模型调用是绕不开的一环。本地跑通时最烦的是 Key 管理ChatBI 内部要调一次模型生成代码客户端本身也要调模型做工具编排如果两边用不同的服务商、不同的 Key排查问题时会非常痛苦。我的做法是统一走 TaoToken 的 API 通道一个 Key 覆盖两个调用点Base URL 和 Model ID 集中配置出问题只看一处日志。官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 注册后到 API Keys 页面拿 Key 即可接入文档里有各语言的示例。下面按「目录结构 → 工具 schema → 本地启动 → 客户端配置 → 验证请求 → 排错」的顺序走一遍每一步都给可复制的代码或配置。2. 目录结构与 TaoToken 统一 Key 的前置准备先把工程骨架搭出来。我用的目录结构如下刻意保持扁平方便你对照排查chatbi-mcp-server/ ├── pandas_mcp_server.py # MCP Server 入口工具注册在这里 ├── chatbi/ │ ├── __init__.py │ ├── data_accessor.py # 数据探查读 Excel/CSV产出 description │ ├── code_generator.py # 调 LLM 生成 pandas 代码 │ └── executor.py # 沙箱执行生成的代码 ├── config.yaml # ACCESS_TOKEN、模型配置 ├── requirements.txt └── data/ └── sales_2024.xlsx # 示例数据requirements.txt内容fastmcp0.4.0 pandas2.0.0 openpyxl3.1.0 openai1.30.0 pydantic2.0 pyyaml6.0这里说明一下为什么用openai这个包TaoToken 的 API 兼容 OpenAI 的请求格式所以直接用官方 SDK把base_url指过去就行不用额外装私有 SDK。config.yaml里集中放两类配置——MCP Server 自身的鉴权 token以及模型调用的通道信息server: access_token: eyJzdWIiOiAidXNlcjEyMyIsICJpYXQiOiAxNzUxODA5ODIwLCAiZXhwIjogMTc1MTgxMzQyMH0 transport: streamable-http host: 0.0.0.0 port: 8000 llm: base_url: https://taotoken.net/api api_key: sk-你的TaoTokenKey model: claude-sonnet-4-5-20250929 timeout: 120关于 Model ID 的选择ChatBI 生成代码这个场景对模型的代码能力要求比较高我实测下来 Claude 系列在 pandas 链式调用和边界处理上更稳遇到「按季度分组再算同比」这类复合需求时不容易漏掉groupby的as_index参数。你可以在模型对话页面先手动试几个 prompt确认模型能稳定输出纯代码块再写进配置。data_accessor.py的核心是产出模型能读懂的「数据描述」。不要直接把整个 DataFrame 塞给模型几万行数据会撑爆上下文。我的做法是只给 schema 加样本import pandas as pd class DataAccessor: def __init__(self, path_or_url: str): if path_or_url.endswith(.csv): self.df pd.read_csv(path_or_url) else: self.df pd.read_excel(path_or_url) self.path path_or_url property def description(self) - str: lines [f数据文件: {self.path}, f总行数: {len(self.df)}, ] lines.append(| 列名 | 类型 | 非空数 | 样例值 |) lines.append(| --- | --- | --- | --- |) for col in self.df.columns: sample self.df[col].dropna().head(3).tolist() lines.append( f| {col} | {self.df[col].dtype} | f{self.df[col].notna().sum()} | {sample} | ) return \n.join(lines)这段description会被get_preview_data直接返回给模型模型据此判断该用哪一列做分组、哪一列做聚合。列名里的中文、空格、特殊符号都要原样保留否则模型生成的代码会KeyError。3. 工具 schema 与可复制配置让模型精准传参MCP 工具最容易踩的坑是「客户端拿不到参数描述」。很多示例代码只写def get_preview_data(path_or_url) - str结果模型看到工具列表时只知道有个叫path_or_url的参数不知道它支持什么格式于是传进来一个.txt路径服务端直接崩。解决办法是用AnnotatedField把类型和描述声明清楚。先写鉴权函数。MCP Server 暴露在本地端口上加一层 Bearer Token 校验能防止同网段的其他程序误调from fastmcp import FastMCP, Context ACCESS_TOKEN eyJzdWIiOiAidXNlcjEyMyIsICJpYXQiOiAxNzUxODA5ODIwLCAiZXhwIjogMTc1MTgxMzQyMH0 def get_bearer_token(ctx: Context) - str: request ctx.get_http_request() authorization_header request.headers.get(Authorization) if not authorization_header: raise ValueError(Authorization header missing) parts authorization_header.split() if len(parts) 2 and parts[0] Bearer and parts[1] ACCESS_TOKEN: return parts[1] raise ValueError(Invalid Authorization header format)然后是工具注册。注意context: Context参数必须放在最后FastMCP 会自动注入不会出现在给模型看的参数列表里from typing import Annotated from pydantic import Field mcp FastMCP(chatbi-mcp-server) mcp.tool( nameget_preview_data, description获取数据文件的列名、类型和样例值用于后续分析前的数据探查 ) def get_preview_data( path_or_url: Annotated[ str, Field(description数据文件路径或URL仅支持Excel(.xlsx)和CSV(.csv)) ], context: Context ) - str: 以AI易读的格式获取数据信息 token get_bearer_token(context) logger.info(fclient token verified: {token[:8]}...) accessor get_data_accessor(path_or_url) return 当前数据信息如下\n accessor.descriptionanalyze_data的参数更多除了数据路径和问题我还加了一个output_format让模型自己决定返回表格还是返回代码mcp.tool( nameanalyze_data, description根据自然语言问题对数据文件进行分析返回统计结果 ) def analyze_data( path_or_url: Annotated[str, Field(description数据文件路径或URL)], question: Annotated[str, Field(description自然语言描述的分析问题)], output_format: Annotated[ str, Field(description输出格式可选 table 或 code默认 table) ] table, context: Context None ) - str: token get_bearer_token(context) accessor get_data_accessor(path_or_url) code generate_code(accessor.description, question) if output_format code: return fpython\n{code}\n result execute_code(accessor.df, code) return result.to_markdown(indexFalse)generate_code内部就是一次标准的 chat completion 调用走 TaoToken 通道from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keyos.getenv(TAOTOKEN_API_KEY), timeout120 ) def generate_code(schema_desc: str, question: str) - str: prompt f你是一个数据分析助手。根据下面的数据描述生成一段 pandas 代码回答问题。 要求 1. 只输出代码不要解释不要 markdown 代码块标记 2. 结果赋值给变量 resultresult 必须是 DataFrame 3. 不要读写文件df 变量已经存在 数据描述 {schema_desc} 问题{question} resp client.chat.completions.create( modelclaude-sonnet-4-5-20250929, messages[{role: user, content: prompt}], temperature0 ) return resp.choices[0].message.content.strip()temperature0是必须的数据分析代码要的是确定性不是创意。execute_code用exec在受限命名空间里跑只放df和pd进去def execute_code(df: pd.DataFrame, code: str) - pd.DataFrame: local_ns {df: df, pd: pd} exec(code, {__builtins__: __builtins__}, local_ns) return local_ns[result]生产环境建议换成子进程隔离这里为了本地跑通先简化。4. 本地启动与验证请求一次自然语言查数启动入口写在pandas_mcp_server.py最下方if __name__ __main__: mcp.run( transportstreamable-http, host0.0.0.0, port8000 )运行python pandas_mcp_server.py看到Uvicorn running on http://0.0.0.0:8000就说明服务起来了。MCP 的 streamable-http 端点默认在/mcp完整地址是http://127.0.0.1:8000/mcp。接下来在客户端里配置。以 Cherry Studio 为例添加 MCP Server 时填配置项值名称chatbi类型streamable-httpURLhttp://127.0.0.1:8000/mcpHeaderAuthorization: Bearer eyJzdWIiOiAidXNlcjEyMyIs...超时300 秒超时一定要设长。ChatBI 一次请求内部要调一次模型生成代码如果代码报错还要重试30 秒根本不够我踩过的坑就是超时设了 60 秒结果模型刚生成完代码连接就断了客户端报local proxy failed排查半天以为是网络问题。保存后切到「工具」标签页如果能看到get_preview_data和analyze_data两个工具且参数描述完整显示说明配置正确。现在做一次完整验证。在对话里输入帮我看看 data/sales_2024.xlsx 这个文件然后按地区统计销售额总和从高到低排序。模型会先调get_preview_data拿到列名假设有地区、销售额、订单日期等列然后调analyze_data传入问题和路径。服务端生成类似这样的代码result df.groupby(地区, as_indexFalse)[销售额].sum().sort_values(销售额, ascendingFalse)执行后返回 Markdown 表格客户端里直接渲染成表格。整个过程你只说了两句话中间的工具调用、代码生成、执行都在本地完成。如果你想跳过客户端直接用 curl 验证服务端是否正常可以发一个初始化请求curl -X POST http://127.0.0.1:8000/mcp \ -H Content-Type: application/json \ -H Authorization: Bearer eyJzdWIiOiAidXNlcjEyMyIs... \ -d {jsonrpc:2.0,id:1,method:tools/list,params:{}}返回的 JSON 里result.tools数组应该包含两个工具定义每个工具的inputSchema里能看到path_or_url的 description。如果 description 是空的回去检查Annotated和Field有没有写对。5. 常见报错排查401、local proxy failed 与 choices 读取失败这一节按真实报错来对。我把跑通过程中遇到的坑列出来你对照日志定位。401 Unauthorized / Invalid Authorization header format服务端返回这个说明get_bearer_token校验没过。三种可能客户端 Header 里没填Authorization填了但格式不是Bearer token比如漏了空格token 值和config.yaml里的access_token不一致。注意parts[0] Bearer是大小写敏感的写成bearer也会失败。排查时在get_bearer_token里把收到的 header 原样打出来一眼就能看出问题。local proxy failed / connection refused客户端连不上服务端。先确认pandas_mcp_server.py还在前台运行没被 CtrlC 掉。然后确认 URL 里的端口和mcp.run里的port一致路径/mcp不能少。如果你在 Docker 里跑服务端host要设成0.0.0.0而不是127.0.0.1否则容器外访问不到。还有一种情况是客户端超时太短服务端还在生成代码客户端已经放弃连接日志里会看到context deadline exceeded把超时调到 300 秒。Error reading choices / choices object has no attribute这个报错来自generate_code里的resp.choices[0]。说明模型调用返回的结构不对。常见原因base_url写成了https://taotoken.net少了/api请求打到了网页端而不是 API 端或者api_key是空的请求被拒后返回了错误结构。检查config.yaml里base_url必须是https://taotoken.net/apiapi_key以sk-开头。另外model字段如果填了一个不存在的 Model ID也会返回错误结构去模型对话页面确认可用的模型名。OAuth 相关报错 / 客户端要求登录有些客户端在添加 MCP Server 时会默认走 OAuth 流程弹出一个授权页面。我们的服务端没有实现 OAuth所以要在客户端配置里选择「无认证」或「Header 认证」模式手动填 Bearer Token。如果客户端强制 OAuth换一个支持自定义 Header 的客户端或者用 Cline、CC Switch 这类配置更灵活的。工具列表为空 / 看不到参数描述服务端起来了但客户端工具页是空的。先确认mcp.tool装饰器有没有漏写name和description是否都填了。然后确认context: Context参数的位置——它必须在参数列表最后如果放在中间FastMCP 可能解析失败。参数描述为空的话检查Annotated[str, Field(description...)]的写法Field要来自pydantic而不是dataclasses。代码执行报 KeyError / 列名对不上模型生成的代码引用了不存在的列。根因通常是description里的列名和实际 DataFrame 不一致比如 Excel 表头有隐藏空格。在DataAccessor.__init__里加一句self.df.columns self.df.columns.str.strip()能解决大部分问题。另外description里给的样例值要足够有代表性如果某列全是空值模型会猜错类型。排错时把日志级别调到 DEBUGgenerate_code里把完整的 prompt 和模型返回都打出来对照着看是哪一步偏了。大部分问题集中在「Key/Base URL 配错」和「参数描述缺失」这两类前者看 401 和 choices 报错后者看工具列表。6. 把这条链路用起来从单次查数到长期编码跑通一次自然语言查数只是起点。真正省时间的是把 ChatBI MCP Server 挂到日常编码流程里——你在 Cursor 里写数据脚本时直接让 Agent 调analyze_data验证中间结果不用切窗口、不用手动跑 pandas。这种长期高频的 Agent 调用场景用 Coding Plan 比按次计费更划算额度覆盖模型调用和工具编排具体可以到 https://taotoken.net/api-keys 看套餐说明。如果你还没决定用哪个模型先去模型对话页面手动试几个数据分析 prompt对比一下生成代码的质量再写进config.yaml。接入文档里有完整的 Base URL、鉴权方式和各语言示例配置卡住时对着查最快https://taotoken.net/doc 。最后给一个实用技巧把data/目录做成软链接指向你真实的业务数据目录这样 MCP Server 不用改配置就能分析新文件。但注意别把生产库直连进来ChatBI 执行的是模型生成的代码本地文件沙箱跑跑可以生产环境务必加隔离层。