AI导哈

SQLMesh MCP服务器

SQLMesh MCP服务器通过模型上下文协议（MCP）为AI代理提供SQLMesh项目的数据转换操作工具，支持模型检查、计划生成、测试审计、列级血缘追踪等功能。

  • by sherman94062
  • Python
  • 本地部署
  • 3 星标
  • 更新于
在 GitHub 查看 sherman94062/sqlmesh-mcp

使用说明

sqlmesh mcp

MCP服务器 SQLMesh --将数据转换操作作为AI代理（Claude、GPT-4和任何兼容MCP的客户端）的工具公开。

这是什么？

sqlmesh mcp 允许AI代理通过以下方式与SQLMesh项目交互 模型上下文协议（MCP）。代理可以检查模型、生成和应用计划、运行测试和审计、探索列级沿袭、管理环境和执行SQL——所有这些都不需要离开AI聊天界面。

这尤其强大，因为SQLMesh的列级沿袭（相对于dbt的一个关键优势）变得可以直接查询：问AI“在哪里 revenue 列在 finance.daily_summary 来自？“并在整个DAG中获得跟踪答案。

工具

工具 说明
sqlmesh_model_list 列出所有带有种类、克戎、粒度和标签的型号
sqlmesh_model_info 单个模型的详细元数据（列、审核、查询预览）
sqlmesh_model_render 为模型渲染完全扩展的SQL
sqlmesh_plan 只读 --预览环境中会发生什么变化
sqlmesh_apply 将计划应用于环境（计划+自动应用）
sqlmesh_run 执行定时间隔到期的计划模型
sqlmesh_test 运行SQLMesh单元测试（基于夹具，无仓库）
sqlmesh_audit 对仓库进行数据质量审核
sqlmesh_dag 遍历模型DAG（上游/下游）
sqlmesh_lineage 列级沿袭 --跨啤酒花追踪柱来源
sqlmesh_environment_list 列出所有环境及其状态
sqlmesh_diff 将环境与产品区分开来
sqlmesh_table_diff 两种环境之间的行级数据差异
sqlmesh_fetchdf 执行SQL并以JSON格式返回结果
sqlmesh_invalidate_environment 使开发/功能环境过期

安装

来源（开发期间推荐）

git clone https://github.com/arthursherman/sqlmesh-mcp
cd sqlmesh-mcp
pip install -e ".[dev]"

PyPI（一旦发布）

pip install sqlmesh-mcp

配置

服务器是通过环境变量配置的：

变量 描述 默认值
SQLMESH_PROJECT_PATH SQLMesh项目的绝对路径 当前工作目录
SQLMESH_GATEWAY 中定义的命名网关 config.yaml （可选） 默认网关

用法

使用克劳德桌面

添加 ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "sqlmesh": {
      "command": "python",
      "args": ["-m", "sqlmesh_mcp.server"],
      "env": {
        "SQLMESH_PROJECT_PATH": "/path/to/your/sqlmesh/project"
      }
    }
  }
}

使用克劳德代码

claude mcp add sqlmesh -- python -m sqlmesh_mcp.server

通过以下方式设置项目路径：

export SQLMESH_PROJECT_PATH=/path/to/your/project

配备MCP检查员（建议开发）

SQLMESH_PROJECT_PATH=/path/to/project \
  npx @modelcontextprotocol/inspector python -m sqlmesh_mcp.server

打开 http://localhost:6274 在浏览器中交互式测试每个工具。

交互示例

一旦连接到Claude或其他MCP客户端，您可以问：

"List all models in my SQLMesh project"
"What columns does db.orders have and what are their types?"
"Show me a plan for the dev environment"
"Where does the revenue column in finance.daily_summary come from?"
"Run tests for the orders model"
"What's the difference between prod and dev right now?"
"Execute: SELECT COUNT(*) FROM db.orders WHERE status = 'complete'"

发展

运行测试

pytest tests/

代码检查

ruff check src/

针对真实SQLMesh项目进行测试

最快的入门方法是SQLMesh内置的DuckDB快速入门：

mkdir sqlmesh-demo && cd sqlmesh-demo
pip install sqlmesh
sqlmesh init duckdb
# Then point this server at it:
SQLMESH_PROJECT_PATH=$(pwd) npx @modelcontextprotocol/inspector python -m sqlmesh_mcp.server

安全注意事项

  • 此服务器使用进程用户的权限执行SQLMesh操作。
  • 这 sqlmesh_apply 和 sqlmesh_run 工具 修改数据 --考虑使用治理层进行包装（请参见 卡斯特兰)如果部署在共享或生产环境中。
  • 这 sqlmesh_fetchdf 该工具执行任意SQL——相应地限制访问。
  • 仅将MCP Inspector代理绑定到localhost；做 不 公开揭露。

路线图

  • []dbt到SQLMesh迁移辅助工具
  • []通过MCP资源暴露语义模型/度量
  • []CI/CD集成（GitHub行动计划摘要）
  • []多项目支持（monorepo）
  • []MCP取样用于代理回填决策

相关项目

许可证

麻省理工学院——见 许可证.

相关 MCP