05_Doris MCP Server：AI时代的数据访问革命

第五篇Doris MCP ServerAI时代的数据访问革命关键字Apache Doris、MCP协议、Model Context Protocol、AI Agent、数据访问、智能交互、LangChain、FastAPI、向量检索标签AI Agent | MCP协议 | 数据库连接 | 智能助手 | 数据基础设施 | Python | FastAPI | 智能运维一、引言当数据库遇见 AI Agent2024 年AI Agent 成为了科技领域最火热的概念。从 AutoGPT 到 Claude Agent从 Cursor 到 Coze无数产品都在试图构建AI 能够自主执行复杂任务的能力。然而当我们真正用这些 Agent 去处理真实业务场景时会发现一个巨大的瓶颈数据访问。一个典型的场景是业务人员问 AI 助手上个月华北区域的销售额是多少环比增长多少AI 需要能够实时查询数据库才能回答这个问题。但在过去这需要为每个数据库定制开发 API工作量巨大且难以维护。MCPModel Context Protocol的出现彻底改变了这个局面。就像 USB-C 统一了设备连接标准一样MCP 正在成为 AI 与数据系统之间的统一接口标准。今天我要为大家介绍 Apache Doris MCP Server——一个让 AI Agent 能够实时、安全、高效地访问 Doris 数据仓库的官方解决方案。二、MCP 协议核心价值2.1 为什么需要 MCP在 MCP 出现之前AI 与数据系统的集成面临多重困境┌─────────────────────────────────────────────────────────────────┐ │ 传统 AI 数据集成的问题 │ │ │ │ 问题1定制开发成本高 │ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │ AI助手 │────│ PostgreSQL │──│ MySQL │────│ Doris │ │ │ └────────┘ └────────┘ └────────┘ └────────┘ │ │ │ │ │ │ │ │ └──► 每个数据库需要独立的 API 适配器 │ │ │ │ 问题2协议不统一 │ │ AI助手需要理解SQL语法、认证方式、连接池管理... │ │ │ │ 问题3安全性难以保障 │ │ 如何控制 AI 能访问哪些数据如何审计 AI 的查询行为 │ └─────────────────────────────────────────────────────────────────┘MCP 的核心理念是为 AI 定义一套标准的数据访问协议就像 HTTP 定义了 web 服务之间的通信方式一样。2.2 MCP 的类比数据世界的 USB-CUSB-C 的成功在于两件事统一的物理接口无论什么设备用同一根线就能连接标准的数据协议设备之间能够互相理解对方的数据格式MCP 对 AI 数据访问的意义完全相同USB-C MCP ─────────────────────────────────────────────────────────────── 物理连接标准 ──────────────────────────► 数据访问协议标准 一根线连接所有设备 ────────────────────► 一个协议访问所有数据库 设备即插即用 ──────────────────────────► AI 即连即用2.3 MCP 协议的核心组成MCP 协议由三部分组成┌─────────────────────────────────────────────────────────────────┐ │ MCP 协议架构 │ │ │ │ ┌──────────────────┐ ┌──────────────────┐ │ │ │ MCP Host │ │ MCP Client │ │ │ │ (Claude Desktop) │◄──────►│ (VSCode Cursor) │ │ │ └────────┬─────────┘ └────────┬─────────┘ │ │ │ │ │ │ │ MCP Protocol │ │ │ └──────────────┬──────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────┐ │ │ │ MCP Server │ │ │ │ (Doris MCP) │ │ │ └────────┬─────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────┐ │ │ │ Data Source │ │ │ │ (Apache Doris) │ │ │ └──────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ MCP Server 提供的核心能力 ├── 工具集ToolsAI 可以调用的函数 ├── 资源ResourcesAI 可以读取的数据 └── 提示Prompts预定义的交互模板三、Doris MCP Server 架构3.1 为什么选择 Doris MCPApache Doris 的 MPP 架构为 MCP Server 提供了独特的优势无主节点架构传统数据库的主节点往往成为 Agent 高频访问的瓶颈。Doris 的无主设计意味着查询可以分散到所有 FE 节点避免单点问题。向量化执行引擎AI Agent 的查询通常是聚合类查询 Doris 的向量化执行可以将这类查询的性能提升 5-10 倍。Multi-Catalog 联邦查询Doris 支持跨数据源的统一查询Agent 可以通过 MCP 同时访问多个数据源无需关心底层的数据位置。3.2 整体架构设计┌─────────────────────────────────────────────────────────────────────────┐ │ Doris MCP Server 架构 │ │ │ │ ┌─────────────────────────────────────────────────────────────────┐ │ │ │ AI Agent / Host │ │ │ │ (Claude Desktop / VSCode Cursor / Coze / FastGPT / LangChain) │ │ │ └─────────────────────────────┬───────────────────────────────────┘ │ │ │ │ │ │ MCP Protocol (JSON-RPC 2.0) │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────────────┐ │ │ │ Doris MCP Server │ │ │ │ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ │ │ │ │ Tools Handler │ │Resources Mgr │ │ Prompts Mgr │ │ │ │ │ └───────┬───────┘ └───────┬───────┘ └───────┬───────┘ │ │ │ │ │ │ │ │ │ │ │ └──────────────────┼──────────────────┘ │ │ │ │ ▼ │ │ │ │ ┌──────────────┐ │ │ │ │ │ Doris Client │ │ │ │ │ │ (pymysql) │ │ │ │ │ └──────┬───────┘ │ │ │ └───────────────────────────┼─────────────────────────────────────┘ │ │ │ │ │ │ MySQL Protocol │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────────────┐ │ │ │ Apache Doris Cluster │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ │ │ FE 1 │◄───►│ FE 2 │◄───►│ FE 3 │ (无主架构) │ │ │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ │ │ │ │ │ │ │ │ └───────────────┼───────────────┘ │ │ │ │ ▼ │ │ │ │ ┌────────────────┐ │ │ │ │ │ BE 节点集群 │ │ │ │ │ │ (向量化执行) │ │ │ │ │ └────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘3.3 核心组件详解Tools Handler工具处理器MCP Server 的核心组件负责处理 AI Agent 的工具调用请求# 伪代码工具调用处理流程classToolsHandler:asyncdefhandle_tool_call(self,tool_name:str,arguments:dict):iftool_namequery:returnawaitself.execute_query(arguments[sql])eliftool_nameexplain_sql:returnawaitself.explain_query(arguments[sql])eliftool_nameget_databases:returnawaitself.list_databases()# ... 其他工具Resources Manager资源管理器管理 AI Agent 可以访问的元数据和资源# 资源定义示例resources[Resource(uridoris://databases,name所有数据库,description返回 Doris 中的所有数据库列表),Resource(uridoris://tables/{database},name数据库表列表,description返回指定数据库中的所有表),Resource(uridoris://columns/{database}/{table},name表结构信息,description返回指定表的列信息)]四、核心工具集与能力4.1 元数据发现工具list_databases()列出所有数据库# MCP 工具调用示例resultawaitmcp_client.call_tool(list_databases)# 返回[analytics, sales, inventory, ml_features]list_tables(database: str)列出数据库中的表resultawaitmcp_client.call_tool(list_tables,arguments{database:sales})# 返回[orders, customers, products, stores]describe_table(database: str, table: str)获取表结构resultawaitmcp_client.call_tool(describe_table,arguments{database:sales,table:orders})# 返回# [# {column: order_id, type: BIGINT, nullable: false},# {column: customer_id, type: BIGINT, nullable: false},# {column: order_date, type: DATETIME, nullable: false},# {column: total_amount, type: DECIMAL(10,2), nullable: false}# ]4.2 SQL 执行与结果格式化query(sql: str)执行 SQL 查询并返回格式化结果resultawaitmcp_client.call_tool(query,arguments{sql: SELECT DATE_FORMAT(order_date, %Y-%m) AS month, COUNT(*) AS order_count, SUM(total_amount) AS total_sales FROM sales.orders WHERE order_date DATE_SUB(NOW(), INTERVAL 6 MONTH) GROUP BY DATE_FORMAT(order_date, %Y-%m) ORDER BY month })# 返回格式化后的查询结果支持 JSON/CSV/Markdownquery_streaming(sql: str)流式执行大查询# 适用于数据量大的查询边查询边返回结果asyncforrowinmcp_client.query_streaming(sql):process_row(row)4.3 性能诊断与优化建议explain_sql(sql: str)获取查询执行计划resultawaitmcp_client.call_tool(explain_sql,arguments{sql:SELECT * FROM large_table WHERE ...})# 返回详细的执行计划用于分析性能瓶颈get_query_profile(query_id: str)获取查询性能分析resultawaitmcp_client.call_tool(get_query_profile,arguments{query_id:f3c8d9e0-...})# 返回查询的各阶段耗时、内存使用、扫描行数等suggest_index(sql: str)索引优化建议增强功能resultawaitmcp_client.call_tool(suggest_index,arguments{sql:SELECT * FROM orders WHERE status pending})# 建议CREATE INDEX idx_status ON orders(status)4.4 审计日志追踪get_access_log(start_time, end_time)获取访问日志resultawaitmcp_client.call_tool(get_access_log,arguments{start_time:2024-01-01 00:00:00,end_time:2024-01-02 00:00:00,limit:100})# 返回访问时间、用户、SQL 语句、执行时长、结果行数get_query_history(user, limit)查询历史resultawaitmcp_client.call_tool(get_query_history,arguments{user:ai_agent,limit:50})五、典型集成场景5.1 FastGPT Doris MCPRAG 知识库构建FastGPT 是一个开源的知识库问答平台支持 RAG 检索增强。将 Doris MCP Server 与 FastGPT 集成后AI 可以直接查询 Doris 中的业务数据来增强回答┌─────────────────────────────────────────────────────────────────┐ │ FastGPT Doris MCP 集成架构 │ │ │ │ 用户提问 │ │ 帮我查一下去年销售额最高的产品是哪个 │ │ │ │ │ │ │ ▼ │ │ ┌───────────────┐ │ │ │ FastGPT │ │ │ │ (RAG 引擎) │ │ │ └───────┬───────┘ │ │ │ │ │ │ 1. 查询 Doris 获取最新销售数据 │ │ ▼ │ │ ┌───────────────┐ │ │ │ Doris MCP │──────────────► Doris Cluster │ │ │ Server │ │ │ │ └───────────────┘ │ │ │ ▼ │ │ ┌───────────────┐ ┌─────────┐ │ │ │ LLM │◄─────────────────────────────│ Doris │ │ │ │ (DeepSeek) │ 结合查询结果生成回答 │ 返回 │ │ │ └───────┬───────┘ └─────────┘ │ │ │ │ │ ▼ │ │ 回答 │ │ 根据销售数据显示2023 年销售额最高的产品是 │ │ XX 型号全年销售额达到 1.2 亿元... │ └─────────────────────────────────────────────────────────────────┘集成配置# FastGPT 配置{model:deepseek-chat,mcpServers:{doris:{command:python,args:[-m,doris_mcp_server],env:{DORIS_HOST:doris-cluster.internal,DORIS_QUERY_PORT:9030,DORIS_USER:ai_agent,DORIS_PASSWORD:xxx}}}}5.2 Coze Doris MCP智能运维工作流在 Coze字节跳动 Bot 平台上创建运维 Bot接入 Doris MCP 实现数据库健康监控和故障诊断┌─────────────────────────────────────────────────────────────────┐ │ Coze Bot Doris MCP 智能运维流程 │ │ │ │ 运维人员智能运维助手 最近数据库有什么异常 │ │ │ │ │ │ │ ▼ │ │ ┌───────────────┐ │ │ │ Coze Bot │ │ │ │ (AI 对话) │ │ │ └───────┬───────┘ │ │ │ │ │ │ MCP Tools 调用 │ │ ├── list_tables() → 检查表数量 │ │ ├── query() → 执行健康检查 SQL │ │ ├── get_query_profile() → 获取慢查询分析 │ │ └── get_access_log() → 检查异常访问 │ │ │ │ ▼ │ │ Doris MCP Server ──► Doris Cluster │ │ │ │ │ │ │ ▼ │ │ 返回诊断结果给 Coze Bot │ │ │ │ │ │ │ ▼ │ │ Bot 回复 │ │ 检测到以下问题 │ │ 1. 表 orders 有 3 个 Tablet 副本不同步 │ │ 2. 过去 1 小时有 5 条慢查询10s │ │ 3. fe01 节点 CPU 使用率 95% │ │ 是否需要我生成优化建议 │ └─────────────────────────────────────────────────────────────────┘5.3 LangChain MCP Adapters 集成LangChain 提供了 MCP 适配器可以轻松将 Doris MCP Server 集成到 LangChain 应用中fromlangchain_mcp_adapters.clientimportMCPClientfromlangchain_openaiimportChatOpenAI# 连接 Doris MCP Servermcp_clientMCPClient(commandpython,args[-m,doris_mcp_server],env{DORIS_HOST:doris.example.com,DORIS_QUERY_PORT:9030,DORIS_USER:ai_agent,DORIS_PASSWORD:xxx})# 创建工具toolsmcp_client.get_tools()# 创建 AgentllmChatOpenAI(modelgpt-4o)agentcreate_react_agent(llm,tools)# 执行查询resultagent.invoke({input:查询过去一周每天的订单量和销售额})六、部署与配置实战6.1 环境要求Doris MCP Server 的环境要求┌─────────────────────────────────────────────────────────────────┐ │ 环境要求 │ │ │ │ Python 版本 3.12 │ │ 依赖包 │ │ ├── fastapi 0.104.0 │ │ ├── uvicorn 0.24.0 │ │ ├── pydantic 2.0.0 │ │ ├── pymysql 1.1.0 │ │ └── python-dotenv 1.0.0 │ │ │ │ 运行环境 │ │ ├── 可以独立部署 │ │ ├── 支持 Docker 部署 │ │ └── 支持 Kubernetes 部署 │ └─────────────────────────────────────────────────────────────────┘6.2 快速安装# 从 GitHub 克隆gitclone https://github.com/apache/doris-mcp-server.gitcddoris-mcp-server# 安装依赖pipinstall-rrequirements.txt# 或者使用 pipx 安装pipxinstalldoris-mcp-server6.3 环境变量配置# 创建 .env 文件cat.envEOF # Doris 连接信息 DORIS_HOSTdoris-cluster.internal DORIS_HTTP_PORT8030 DORIS_QUERY_PORT9030 DORIS_USERai_agent DORIS_PASSWORDyour_password_here # 可选TLS 连接 DORIS_SSL_ENABLEDfalse DORIS_SSL_CERT_PATH/path/to/cert.pem # 可选连接池配置 DORIS_MAX_CONNECTIONS10 DORIS_CONNECTION_TIMEOUT30 # 可选日志级别 LOG_LEVELINFO LOG_FILE/var/log/doris-mcp-server.log EOF6.4 服务启动与验证独立启动# 启动服务python-mdoris_mcp_server# 或者使用 uvicornuvicorn doris_mcp_server.main:app--host0.0.0.0--port8000Docker 部署# 构建镜像dockerbuild-tdoris-mcp-server:latest.# 运行容器dockerrun-d\--namedoris-mcp\-p8000:8000\--env-file .env\doris-mcp-server:latestKubernetes 部署# deployment.yamlapiVersion:apps/v1kind:Deploymentmetadata:name:doris-mcp-serverspec:replicas:2selector:matchLabels:app:doris-mcptemplate:metadata:labels:app:doris-mcpspec:containers:-name:doris-mcpimage:doris-mcp-server:latestports:-containerPort:8000envFrom:-secretRef:name:doris-credentials---apiVersion:v1kind:Secretmetadata:name:doris-credentialstype:OpaquestringData:DORIS_HOST:doris-cluster.internalDORIS_QUERY_PORT:9030DORIS_USER:ai_agentDORIS_PASSWORD:xxx6.5 服务验证# 健康检查curlhttp://localhost:8000/health# 测试查询工具curl-XPOST http://localhost:8000/mcp/tools/query\-HContent-Type: application/json\-d{sql: SELECT 1 as test}# 查看可用工具列表curlhttp://localhost:8000/mcp/tools七、安全配置与最佳实践7.1 权限隔离为 AI Agent 创建专用的数据库用户-- 创建只读用户推荐CREATEUSERai_readonly%IDENTIFIEDBYstrong_password;GRANTSELECTON*.*TOai_readonly%;-- 创建分析用户可访问特定数据库CREATEUSERai_analytics%IDENTIFIEDBYstrong_password;GRANTSELECTONanalytics.*TOai_analytics%;GRANTSELECTONsales.*TOai_analytics%;-- 禁止危险操作REVOKEINSERT,UPDATE,DELETE,DROPON*.*FROMai_readonly%;7.2 SQL 审计启用 SQL 审计日志-- 查看审计日志SELECTquery_id,user,sql,start_time,end_time,scan_rowsFROMDoris_Query_Profile.__internal_query_profile__WHEREuserai_agentORDERBYstart_timeDESCLIMIT100;7.3 限流配置防止 AI Agent 的异常查询拖垮数据库-- 创建工作负载组CREATEWORKLOADGROUPai_queriesPROPERTIES(cpu_share1024,memory_limit20%,max_concurrency10,max_queue_size100,query_timeout300-- 5 分钟超时);7.4 敏感数据脱敏MCP Server 层面实现数据脱敏# 配置脱敏规则SENSITIVE_COLUMNS{users:[password,id_card,phone],orders:[credit_card_number]}app.middlewareasyncdefmask_sensitive_data(request:Request,call_next):responseawaitcall_next(request)# 实现脱敏逻辑returnmask_response(response,SENSITIVE_COLUMNS)八、性能优化与监控8.1 连接池配置# doris_mcp_server/config.pyfromsqlalchemy.poolimportQueuePool enginecreate_engine(mysqlpymysql://...,poolclassQueuePool,pool_size10,max_overflow20,pool_pre_pingTrue,pool_recycle3600)8.2 查询缓存fromfunctoolsimportlru_cacheimporthashliblru_cache(maxsize1000)defcached_query(sql:str)-dict:缓存高频查询结果returnexecute_query(sql)defget_query_cache_key(sql:str)-str:生成查询缓存键returnhashlib.md5(sql.encode()).hexdigest()8.3 监控指标集成 Prometheus 监控fromprometheus_clientimportCounter,Histogram,Gauge# 定义指标query_countCounter(doris_mcp_queries_total,Total queries,[status])query_durationHistogram(doris_mcp_query_duration_seconds,Query duration)active_connectionsGauge(doris_mcp_active_connections,Active connections)app.middlewareasyncdeftrack_metrics(request:Request,call_next):query_count.labels(statussuccess).inc()withquery_duration.time():responseawaitcall_next(request)returnresponse九、常见问题与排查9.1 连接超时症状Connection timeout after 30s排查步骤# 1. 检查 Doris FE 是否可达telnet doris-fe9030# 2. 检查用户名密码mysql-hdoris-fe-P9030-uai_agent-p# 3. 检查 MCP Server 日志tail-f/var/log/doris-mcp-server.log9.2 权限不足症状Access denied for user解决方案-- 检查用户权限SHOWGRANTSFORai_agent%;-- 重新授权GRANTSELECTON*.*TOai_agent%;FLUSHPRIVILEGES;9.3 查询超时症状Query timeout after 300s优化建议启用查询缓存添加合适的索引调整查询超时配置使用 LIMIT 限制返回行数十、总结与展望Doris MCP Server 是 Apache Doris 在 AI 时代的重要布局。它通过标准化 MCP 协议让 AI Agent 能够实时、安全地访问 Doris 数据仓库解决了 AI 数据集成领域的多重困境。对于企业而言Doris MCP Server 带来了三重价值降低集成成本一个协议连接所有 AI 应用无需为每个应用单独开发 API提升数据时效性AI 助手可以实时查询最新数据回答更准确保障数据安全通过权限控制、审计日志、限流配置确保 AI 的数据访问安全可控展望未来我期待看到以下演进更多工具支持数据导入、备份恢复、Schema 变更等运维工具流式响应支持 SseServerStream 实时推送查询结果多数据库支持不仅支持 Doris还能连接其他数据源作为联邦查询入口智能优化基于历史查询自动分析性能瓶颈并提供优化建议AI Agent 的时代已经来临而 Doris MCP Server 正是那个让 AI 真正理解企业数据的桥梁。