Apache Doris + MCP：Agent时代的实时数据分析底座

1. 为什么 Agent 需要 Apache Doris 做实时数据分析底座

Agent 要真正干活，绕不开一个动作：查数据。用户问「昨天华东区退货率最高的三个 SKU 是什么」，Agent 得把这句话翻译成 SQL，打到某个能秒回明细和聚合的库上，再把结果组织成人话。问题在于，大多数团队给 Agent 接的是业务库或离线数仓——业务库扛不住高频聚合，离线数仓延迟按小时算，Agent 等不起。

Apache Doris 在这里的角色就很清楚了：它是一款 MPP 架构的实时分析型数据库，写入即可查，明细和聚合走同一套 SQL 引擎，单表亿级数据做过滤聚合通常在百毫秒级返回。对 Agent 来说，这意味着「查一下」不再是异步任务，而是可以塞进对话循环里的同步动作。MCP（Model Context Protocol）则是把这层能力标准化暴露给 Agent 的协议层——你不用给每个 Agent 框架写一套适配，只要起一个 MCP Server，把 Doris 的查询能力注册成工具，Claude Desktop、Cline、Codex 这类客户端就能直接调用。

这套组合适合谁？一是做数据分析类 Agent 的团队，需要 Agent 直接查明细和指标；二是已经有 Doris 集群、想让 AI 工作流复用现有数据的团队；三是想快速验证「Agent + 实时数仓」可行性、不想先造一套 API 网关的开发者。下面我从环境准备讲到可复制的 config.toml，再到一次真实的 Agent 查询验证，把踩过的坑一并写出来。

2. TaoToken 前置准备：给 Agent 配一个稳定的模型入口

MCP Server 负责查 Doris，但 Agent 本身需要一个能调用工具的大模型。这里我用 TaoToken 作为模型接入层，它提供 OpenAI 兼容接口，Claude Code、Cline、Codex 这些支持 MCP 的客户端都能直接对接。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。

先说清楚三件套，这是后面所有配置的基础：

拿 Key 的路径：打开 https://taotoken.net/api-keys ，登录后点创建，复制保存。如果你用的是 Claude Code，接入文档在 https://taotoken.net/doc ，里面有针对 Anthropic 协议的说明；Cline 或 Cursor 这类走 OpenAI 协议的，直接把 Base URL 和 Key 填进设置即可。

这里有个容易混的点：MCP Server 和模型入口是两条独立的链路。MCP Server 跑在本地或内网，负责连 Doris；模型入口走 TaoToken，负责推理和工具调用决策。Agent 客户端同时配置这两者——模型告诉它「该调哪个工具」，MCP Server 真正去执行查询。所以你在 TaoToken 控制台创建 Key 之后，还要在客户端里把 MCP Server 注册进去，两者缺一不可。

如果你打算长期跑编码或 Agent 任务，可以考虑 Coding Plan，额度更划算，入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。验证模型是否通，用模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 发一条消息即可，确认返回正常再往下走。

3. 可复制配置：Doris MCP Server 的 config.toml 骨架

这一节是核心。MCP Server 我用 Python + FastMCP 写，通信模式选 stdio（本地低延迟，客户端直接拉起进程）。先装依赖：

pip install fastmcp pymysql

Doris 走 MySQL 协议，所以用 pymysql 连接即可，不需要额外驱动。下面是 config.toml 骨架，路径放在项目根目录doris-mcp/config.toml：

[server] name = "doris-mcp" version = "0.1.0" transport = "stdio" [doris] host = "127.0.0.1" port = 9030 user = "root" password = "" database = "demo" charset = "utf8mb4" connect_timeout = 5 read_timeout = 30 [limits] max_rows = 500 allow_write = false query_timeout_ms = 10000 [tools] enable_query = true enable_schema = true enable_metrics = true

几个参数说明。port = 9030是 Doris FE 的 MySQL 协议端口，不是 8030 的 HTTP 端口，填错会一直连不上。max_rows限制单次返回行数，防止 Agent 一句SELECT *把上下文撑爆。allow_write = false是硬开关，Agent 场景默认只读，避免误删误改。query_timeout_ms配合 Doris 的 query_timeout 会话变量一起用。

接着是 Server 主文件doris-mcp/server.py，把连接和工具注册写全：

import tomllib import pymysql from fastmcp import FastMCP with open("config.toml", "rb") as f: cfg = tomllib.load(f) mcp = FastMCP(cfg["server"]["name"]) def get_conn(): d = cfg["doris"] return pymysql.connect( host=d["host"], port=d["port"], user=d["user"], password=d["password"], database=d["database"], charset=d["charset"], connect_timeout=d["connect_timeout"], read_timeout=d["read_timeout"], cursorclass=pymysql.cursors.DictCursor, ) @mcp.tool() def run_query(sql: str) -> list: """执行只读 SQL，返回明细或聚合结果。""" if not cfg["limits"]["allow_write"] and not sql.strip().lower().startswith("select"): return [{"error": "only SELECT is allowed"}] conn = get_conn() try: with conn.cursor() as cur: cur.execute("SET query_timeout = %s", (cfg["limits"]["query_timeout_ms"] // 1000,)) cur.execute(sql) rows = cur.fetchmany(cfg["limits"]["max_rows"]) return rows finally: conn.close() @mcp.tool() def list_tables() -> list: """列出当前库的所有表，供 Agent 探索 schema。""" conn = get_conn() try: with conn.cursor() as cur: cur.execute("SHOW TABLES") return cur.fetchall() finally: conn.close() if __name__ == "__main__": mcp.run(transport=cfg["server"]["transport"])

工具注册的关键是 docstring——MCP 客户端会把 docstring 作为工具描述喂给模型，写清楚「只读」「返回明细或聚合」能显著降低模型乱调的概率。list_tables这个工具别省，Agent 第一次接触你的库时不知道有哪些表，给它一个探索入口比在 prompt 里硬编码表名灵活得多。

然后在客户端侧注册这个 Server。以 Cline 为例，在 MCP 设置里加：

{ "mcpServers": { "doris-mcp": { "command": "python", "args": ["/path/to/doris-mcp/server.py"], "env": {} } } }

Claude Code 的配置在~/.claude/settings.json或项目级.mcp.json，结构类似，把 command 和 args 指向上面的 server.py 即可。Codex 用户如果走 auth.json 体系，模型侧填 TaoToken 的 Base URL 和 Key，MCP 侧同样用上面的 JSON 结构注册。

4. 验证请求：让 Agent 发起一次真实的 Doris 实时查询

配置写完，先别急着开 Agent，手动验证 Server 能跑通。直接命令行启动：

cd doris-mcp && python server.py

stdio 模式下它不会打印监听端口，而是等待标准输入。更直观的验证是写个测试脚本，模拟一次工具调用：

import asyncio from fastmcp import Client async def main(): async with Client("doris-mcp/server.py") as client: tools = await client.list_tools() print("registered:", [t.name for t in tools]) res = await client.call_tool("run_query", { "sql": "SELECT region, SUM(amount) AS gmv FROM orders WHERE dt='2025-06-22' GROUP BY region ORDER BY gmv DESC" }) print(res) asyncio.run(main())

跑通的话你会看到registered: ['run_query', 'list_tables']，以及按 region 聚合的 GMV 结果。这一步确认了 Server 到 Doris 的链路没问题。

接下来才是真正的 Agent 验证。在 Cline 里把模型配成 TaoToken 的端点，然后发一句自然语言：

帮我查一下 2025-06-22 各区域的 GMV，按从高到低排。

正常流程是：模型识别出需要调run_query，生成 SQL，MCP Server 执行后把结果回传，模型再组织成表格。实测下来，从发问到拿到结果通常在 2 到 4 秒，其中 Doris 查询本身占 100 到 300 毫秒，剩下是模型推理和工具往返。如果你看到 Agent 回复里带了具体数字和区域名，说明整条链路通了。

想验证聚合之外的明细查询，可以再问「列出退货率最高的 5 个 SKU 及其退货原因」，观察它是否会先调list_tables探索，再拼多表关联。这一步能暴露 schema 描述是否清晰——如果模型拼错表名，回去补 docstring 或加一个describe_table工具。

5. 本篇常见错排查：401、local proxy failed 与 reading choices

配置过程中最容易撞的几个报错，我按实际遇到的频率排一下。

401 Unauthorized。两种可能：一是 TaoToken 的 API Key 没填对或过期，去 https://taotoken.net/api-keys 重新生成；二是 Key 填了但 Base URL 写成了https://taotoken.net/api/v1导致路径重复。OpenAI 兼容客户端一般自己补/v1，Base URL 填https://taotoken.net/api即可。检查方法是直接用 curl 打一发：

curl https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer sk-你的key" \ -H "Content-Type: application/json" \ -d '{"model":"你的模型ID","messages":[{"role":"user","content":"ping"}]}'

返回 200 说明模型侧没问题，401 就回去查 Key。

local proxy failed。这个报错通常出现在客户端试图连 MCP Server 时。原因多是 server.py 路径写错，或者 Python 环境里没装 fastmcp。先在终端手动python /path/to/server.py跑一遍，能起来再填进客户端配置。另外 stdio 模式下客户端会自己拉起进程，如果你同时在终端手动跑着，端口或进程会冲突，记得先关掉手动进程。

reading choices 相关报错。这类多半是模型返回格式不符合客户端预期，常见于模型 ID 填错或用了不兼容的模型。确认 TaoToken 控制台里该模型 ID 拼写一致，并且客户端选的是 OpenAI 兼容模式而非 Anthropic 原生模式（除非你确实走 Anthropic 协议）。Claude Code 走 Anthropic 协议时，接入文档 https://taotoken.net/doc 里有专门的 Base URL 写法，别和 OpenAI 模式混用。

Doris 连接超时。报(2003, "Can't connect to MySQL server")时，先确认 9030 端口通不通：telnet 127.0.0.1 9030。Doris 的 FE 可能有多个节点，如果配的是负载地址，确认 VIP 或域名解析正常。另外read_timeout设太小，大聚合查询会被掐断，按 config.toml 里 30 秒起步。

Agent 不调工具、直接编答案。这不是报错但更隐蔽。原因是工具 docstring 太模糊，模型不知道什么时候该用。把run_query的描述改成「当用户询问具体数据、指标、明细时调用，禁止凭记忆回答」，并在系统提示里强调「所有数据问题必须走 doris-mcp」。实测这一句能明显提升工具调用率。

6. 把实时分析接进 Agent 工作流的下一步

链路跑通之后，真正决定好不好用的是查询治理。Agent 生成的 SQL 不可预测，我一般会在 Server 侧加一层轻量校验：拦截没有LIMIT的明细查询、拦截SELECT *、对高频表建好物化视图。Doris 的物化视图对聚合类查询提速很明显，Agent 反复问「今日 GMV」这类指标时，命中物化视图能压到几十毫秒。

另一个实用技巧是把常用指标固化成工具，而不是让模型每次现拼 SQL。比如注册一个get_daily_gmv(date)工具，内部写死聚合逻辑，模型只负责传日期。这样既降低出错率，也避免模型在 SQL 里写出全表扫描。工具数量控制在 10 个以内，太多反而让模型选择困难。

模型入口这边，长期跑 Agent 任务建议用 Coding Plan，额度和稳定性比按次调用更省心，入口 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。需要调试模型行为时，模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 可以快速试 prompt。API Key 管理在 https://taotoken.net/api-keys ，接入细节看 https://taotoken.net/doc 。

最后提醒一句：Doris 的 MCP Server 别直接暴露到公网，stdio 本地模式最稳；如果必须走 SSE，至少加一层鉴权，并且allow_write永远保持 false。Agent 再聪明，也不该拿到你生产库的写权限。