1. 为什么 Agent 直连数仓总在“最后一公里”翻车Apache Doris 在实时数据分析圈子里已经不算新面孔MPP 架构、列式存储、Merge-on-Write 这些特性让它在宽表聚合和秒级写入场景里表现稳定。但真正把 Agent 接进来的时候很多人会发现一个尴尬的现实模型能写 SQL却不知道怎么安全地拿到表结构能生成查询却没法把结果稳定地回传给上层应用。MCPModel Context Protocol要解决的正是这段“最后一公里”——它把数据库能力包装成模型可发现、可调用的工具让 Agent 不用硬编码就能查询 Doris。这篇文章面向的是想让智能体直接查询数仓的开发者。我会从实际接入的角度出发给出 MCP Server 连接 Doris 的可复制配置、Agent 调用链路的完整示例以及查询延迟和结果正确性的验证动作。你不需要先成为 MCP 协议专家只要手里有一个能跑的 Doris 实例和基本的 Python 环境就能跟着把这条链路搭起来。核心检索词先明确Apache Doris 提供实时分析底座MCP 提供标准化工具接口Agent 负责编排调用。三者组合起来才是可复现的智能化分析底座。适合谁做数据平台的后端、做 AI 应用的工程师、以及想把 BI 查询能力开放给智能体的团队。不适合谁只想跑单条 SQL 看看结果、不打算做工具封装的场景直接用 MySQL 客户端更省事。我试过把 Doris 的查询接口直接暴露给模型结果模型经常编造字段名或者在聚合函数上出错。后来换成 MCP 封装把表结构和常用查询模板作为资源暴露出去错误率明显下降。这不是模型变聪明了而是约束变清晰了。2. TaoToken 前置把模型调用和工具调用拆开看在搭 Doris MCP 之前需要先把模型侧的调用通道准备好。很多教程会把模型接入和工具接入混在一起讲导致排障时分不清是模型没响应还是 MCP Server 没起来。我的建议是拆成两层模型层用 TaoToken 提供的统一接口工具层用本地或远程的 MCP Server。TaoToken 的 API 地址是 https://taotoken.net/api它兼容常见的模型调用格式你可以在模型对话页面先验证模型是否能正常返回。这一步的意义在于当 Agent 调用 Doris 失败时你能快速判断是模型侧的问题还是工具侧的问题。如果模型对话都返回不了那后面 MCP 的调试就是白费力气。具体操作上先到官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 了解接入方式然后在控制台创建 API Key。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。拿到 Key 之后先别急着写 Agent用模型对话页面发一条简单消息确认通道是通的。这里有个容易踩的坑有人把 MCP Server 的地址填成模型 API 的地址结果两边都报错。记住TaoToken 的 API 是给模型用的MCP Server 是你自己部署的、用来连接 Doris 的服务两者是独立的。Agent 的工作是把模型输出转换成 MCP 工具调用再把工具返回的结果喂回模型。如果你打算长期跑编码类或 Agent 类任务可以关注 Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite它更适合持续性的开发场景。但本文的重点是 Doris 查询链路模型侧只要保证能稳定调用即可。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite遇到请求格式问题可以先查这里。Claude Code 相关的接入在 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite如果你用 Claude Code 作为 Agent 宿主可以参考这个页面配置。3. 可复制配置Doris MCP Server 的 settings 与工具定义这一节给出可以直接复制修改的配置片段。先说明目录结构避免路径对不上。假设你的工作目录是~/doris-mcp里面放一个server.py和一个settings.json。Doris 的连接信息通过环境变量注入不要硬编码在代码里。先看settings.json这是 MCP Server 的配置文件路径与原文一致{ mcpServers: { doris-analytics: { command: python, args: [/home/yourname/doris-mcp/server.py], env: { DORIS_HOST: 127.0.0.1, DORIS_PORT: 9030, DORIS_USER: root, DORIS_PASSWORD: , DORIS_DATABASE: analytics, MCP_SERVER_PORT: 8321 } } } }这个文件的作用是告诉 Agent 宿主有一个叫doris-analytics的 MCP Server用 Python 启动环境变量里带着 Doris 的连接信息。注意args里的路径要换成你机器上的真实路径Windows 下用反斜杠或双引号转义。接下来是server.py的核心部分。这里用官方的 MCP Python SDK 来定义工具工具名是doris_query参数是一个 SQL 字符串。为了让模型知道有哪些表可用再定义一个资源doris://schema返回表结构摘要。import os import pymysql from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent, Resource app Server(doris-analytics) def get_conn(): return pymysql.connect( hostos.environ[DORIS_HOST], portint(os.environ[DORIS_PORT]), useros.environ[DORIS_USER], passwordos.environ[DORIS_PASSWORD], databaseos.environ[DORIS_DATABASE], charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, ) app.list_tools() async def list_tools(): return [ Tool( namedoris_query, description在 Apache Doris 上执行只读 SQL 查询返回 JSON 结果, inputSchema{ type: object, properties: { sql: {type: string, description: SELECT 语句} }, required: [sql], }, ) ] app.call_tool() async def call_tool(name, arguments): if name ! doris_query: raise ValueError(funknown tool: {name}) sql arguments[sql].strip() if not sql.lower().startswith(select): return [TextContent(typetext, text仅允许 SELECT 查询)] conn get_conn() try: with conn.cursor() as cur: cur.execute(sql) rows cur.fetchall() return [TextContent(typetext, textstr(rows))] finally: conn.close() app.list_resources() async def list_resources(): return [Resource(uridoris://schema, nameDoris 表结构, mimeTypetext/plain)] app.read_resource() async def read_resource(uri): if str(uri) ! doris://schema: raise ValueError(unknown resource) conn get_conn() try: with conn.cursor() as cur: cur.execute(SHOW TABLES) tables [list(r.values())[0] for r in cur.fetchall()] lines [f数据库 {os.environ[DORIS_DATABASE]} 下的表] lines.extend(tables) return \n.join(lines) finally: conn.close() async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这段代码里有两个关键点。第一工具只允许 SELECT避免 Agent 误删数据。第二资源doris://schema让模型在生成 SQL 前能先拉取表名减少编造字段的概率。实际部署时你可以把SHOW TABLES换成查询information_schema.columns返回字段名和类型效果更好。依赖安装命令pip install mcp pymysql如果你用的是 Cline 或 CC Switch 这类支持 MCP 的客户端把settings.json放到对应的配置目录即可。Cline 的 MCP 配置通常在设置界面里粘贴 JSONCC Switch 则读取本地配置文件。无论哪种Base URL、Key、Model ID 三件套要写全Base URL 用https://taotoken.net/apiKey 用控制台生成的Model ID 按你选的模型填。这三项缺一个Agent 就调不起来。4. 验证请求从 Agent 发起到 Doris 返回的完整链路配置写完之后不要直接上复杂查询先用一条最简单的 SQL 验证链路。启动 MCP Server 后在 Agent 宿主里发一条消息“帮我查一下 analytics 库里的表”。Agent 应该先调用doris://schema资源拿到表名列表再决定是否执行查询。如果你想手动验证 MCP Server 是否正常可以用 stdio 方式直接调用。下面是一个最小客户端示例import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython, args[/home/yourname/doris-mcp/server.py], env{ DORIS_HOST: 127.0.0.1, DORIS_PORT: 9030, DORIS_USER: root, DORIS_PASSWORD: , DORIS_DATABASE: analytics, }, ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) result await session.call_tool( doris_query, {sql: SELECT COUNT(*) AS cnt FROM orders} ) print(查询结果:, result.content[0].text) asyncio.run(main())运行后如果看到类似查询结果: [{cnt: 12843}]的输出说明 Doris 连接和 MCP 工具调用都正常。这一步的延迟主要来自 Doris 查询本身MCP 层的开销通常在毫秒级。你可以用time命令包一下记录从调用到返回的耗时。在 Agent 侧调用链路是这样的用户提问 → 模型判断需要查数 → 模型输出工具调用请求 → Agent 宿主转发给 MCP Server → MCP Server 执行 SQL → 结果回传模型 → 模型生成自然语言回答。任何一环断了表现都是“Agent 没反应”或“回答里没有数据”。所以验证时要分段确认先确认模型能回消息再确认 MCP Server 能列工具最后确认 SQL 能返回结果。结果正确性方面建议拿一条你已知答案的 SQL 做对照。比如SELECT COUNT(*) FROM orders WHERE dt 2024-01-01先在 MySQL 客户端里跑一遍记下数字再让 Agent 跑同样的查询。如果数字一致说明链路没有丢数据或改条件。如果 Agent 返回的数字不对优先检查模型是否改写了 SQL可以在 MCP Server 里加日志把实际执行的 SQL 打印出来。5. 常见错排查401、local proxy failed 与 reading choices这一节对照真实报错给出排查路径。第一个高频错误是401 Unauthorized。如果你在 Agent 侧看到这个通常是 TaoToken 的 API Key 没填对或过期了。检查settings.json或宿主配置里的 Key 是否和控制台一致注意不要有多余空格。如果 MCP Server 侧报 401那是 Doris 的账号密码问题检查DORIS_USER和DORIS_PASSWORD。第二个错误是local proxy failed。这个报错通常出现在 Agent 宿主尝试连接 MCP Server 时原因可能是server.py路径写错、Python 环境不对、或者端口被占用。排查步骤先在终端手动运行python /home/yourname/doris-mcp/server.py看是否报错再用ps aux | grep server.py确认进程是否活着最后检查settings.json里的args路径是否和实际一致。Windows 下路径要用双反斜杠或正斜杠。第三个错误是reading choices相关的解析失败。这通常发生在模型返回格式不符合预期时比如模型把工具调用写成了普通文本或者返回了多个 choice 但 Agent 只解析了第一个。解决方法是检查模型输出日志确认工具调用的 JSON 结构是否完整。如果用的是 TaoToken 的模型对话接口可以先在模型对话页面测试同样的提示词看返回是否稳定。第四个错误是 OAuth 相关的认证失败。MCP 协议支持 OAuth 2.0但本地开发时通常用 stdio 或简单 token 就够了。如果你在远程 MCP Server 上启用了 OAuth确保客户端配置里的 token 端点、client_id、client_secret 都正确。本地场景下建议先用 stdio 跑通再考虑远程部署。还有一个隐蔽的坑Doris 的SHOW TABLES返回的字段名在不同版本里可能不一样有的返回Tables_in_analytics有的返回Table_name。如果你的资源读取代码写死了字段名换版本就会报 KeyError。用list(r.values())[0]取第一个值更稳妥。排查时记住一个原则先隔离再定位。把模型调用和 MCP 调用分开测把 Doris 查询和 MCP 封装分开测。哪一层单独跑不通问题就在哪一层。不要一上来就怀疑协议不兼容大多数问题都是配置路径或认证信息写错了。6. 把这条链路用起来从验证到日常查询链路跑通之后你可以把常用的查询封装成更具体的工具而不是只暴露一个通用doris_query。比如定义doris_daily_gmv参数是日期内部固定 SQL 模板。这样模型不需要生成复杂 SQL只需要填参数出错概率更低。工具描述里写清楚每个参数的格式和含义模型就能更准确地调用。对于长期运行的 Agent 场景建议把 MCP Server 做成常驻进程而不是每次调用都启动。可以用 systemd 或 supervisor 管理日志输出到文件方便排查。Doris 侧则要注意连接池和查询超时避免 Agent 发起大查询把连接占满。在pymysql.connect里加上read_timeout和connect_timeout给查询设一个上限。如果你需要更完整的接入示例和配置模板可以到接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 查看。模型对话验证在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewriteAPI Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。长期编码和 Agent 任务可以看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite。最后说一个实用技巧在 MCP Server 里加一个查询耗时统计每次调用后把 SQL 和执行时间写到日志。跑一段时间后你会清楚哪些查询是 Agent 高频发起的哪些表被扫得最多。这些数据反过来可以指导你优化 Doris 的索引和分桶让整条链路越跑越快。