企业官网建设流程全解析

1. 为什么我要自己写一个 ChatBI MCP Server

先说清楚这东西是什么。ChatBI MCP Server 是一个把「自然语言查数」能力封装成标准工具协议的服务端程序，它让 Claude、Cursor、Cherry Studio 这类支持 MCP 的客户端，能够通过一次对话直接读取你的 Excel/CSV 文件、生成分析代码、跑出统计结果。适合谁？适合手头有一堆业务表格、又不想每次都写 pandas 脚本的开发者，也适合正在学 Agent 工具调用、想找一个能完整跑通的实战项目的人。

我试过直接让大模型读 CSV 然后口算，数据量一过千行就开始胡说，列名稍微复杂点就张冠李戴。原因很简单：模型没有真正执行代码，它在做概率补全。ChatBI 的思路是把「生成代码」和「执行代码」分开——模型只负责根据数据描述写 pandas 代码，真正的计算交给本地 Python 进程。这样准确率是数量级的提升。

MCP（Model Context Protocol）在这里扮演的角色是「工具插座」。你不需要为每个客户端写一套适配层，只要按协议暴露工具，任何支持 MCP 的客户端都能即插即用。我们要实现两个核心工具：get_preview_data负责数据探查，把列名、类型、前几行样本以 Markdown 形式返回给模型；analyze_data负责接收自然语言问题，内部调用大模型生成代码并执行，返回结果表格。

整条链路里，大模型调用是绕不开的一环。本地跑通时最烦的是 Key 管理：ChatBI 内部要调一次模型生成代码，客户端本身也要调模型做工具编排，如果两边用不同的服务商、不同的 Key，排查问题时会非常痛苦。我的做法是统一走 TaoToken 的 API 通道，一个 Key 覆盖两个调用点，Base URL 和 Model ID 集中配置，出问题只看一处日志。官网在 https://taotoken.net/?utm_source=taotoken_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内容：

fastmcp>=0.4.0 pandas>=2.0.0 openpyxl>=3.1.0 openai>=1.30.0 pydantic>=2.0 pyyaml>=6.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路径，服务端直接崩。解决办法是用Annotated+Field把类型和描述声明清楚。

先写鉴权函数。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( name="get_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(f"client token verified: {token[:8]}...") accessor = get_data_accessor(path_or_url) return "当前数据信息如下：\n" + accessor.description

analyze_data的参数更多，除了数据路径和问题，我还加了一个output_format，让模型自己决定返回表格还是返回代码：

@mcp.tool( name="analyze_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 f"```python\n{code}\n```" result = execute_code(accessor.df, code) return result.to_markdown(index=False)

generate_code内部就是一次标准的 chat completion 调用，走 TaoToken 通道：

from openai import OpenAI client = OpenAI( base_url="https://taotoken.net/api", api_key=os.getenv("TAOTOKEN_API_KEY"), timeout=120 ) def generate_code(schema_desc: str, question: str) -> str: prompt = f"""你是一个数据分析助手。根据下面的数据描述，生成一段 pandas 代码回答问题。 要求： 1. 只输出代码，不要解释，不要 markdown 代码块标记 2. 结果赋值给变量 result，result 必须是 DataFrame 3. 不要读写文件，df 变量已经存在 数据描述： {schema_desc} 问题：{question} """ resp = client.chat.completions.create( model="claude-sonnet-4-5-20250929", messages=[{"role": "user", "content": prompt}], temperature=0 ) return resp.choices[0].message.content.strip()

temperature=0是必须的，数据分析代码要的是确定性，不是创意。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( transport="streamable-http", host="0.0.0.0", port=8000 )

运行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 一次请求内部要调一次模型生成代码，如果代码报错还要重试，30 秒根本不够，我踩过的坑就是超时设了 60 秒，结果模型刚生成完代码连接就断了，客户端报local proxy failed，排查半天以为是网络问题。

保存后切到「工具」标签页，如果能看到get_preview_data和analyze_data两个工具，且参数描述完整显示，说明配置正确。

现在做一次完整验证。在对话里输入：

帮我看看 data/sales_2024.xlsx 这个文件，然后按地区统计销售额总和，从高到低排序。

模型会先调get_preview_data，拿到列名（假设有地区、销售额、订单日期等列），然后调analyze_data，传入问题和路径。服务端生成类似这样的代码：

result = df.groupby("地区", as_index=False)["销售额"].sum().sort_values("销售额", ascending=False)

执行后返回 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还在前台运行，没被 Ctrl+C 掉。然后确认 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/api，api_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里给的样例值要足够有代表性，如果某列全是空值，模型会猜错类型。

排错时把日志级别调到 DEBUG，generate_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 执行的是模型生成的代码，本地文件沙箱跑跑可以，生产环境务必加隔离层。