包装 Dagster GraphQL API 的 MCP 服务器——从任何 MCP 客户端监控和运行运行运行,资产,调度,传感器和回填
概览
一个MCP服务器,它让AI代理 完全能见度和控制您的 Dagster 实例—— 就像一个用于您数据管道的 SRE. 与任何MCP客户端合作:克洛德代码,克洛德桌面,Cursor等. Dagster正式开发并支持他们自己的Dagster Plus MCP服务器. 见这里的文件。 数据管道在凌晨3点中断。 时间表静静地停止开火。 资产沉闷。 而不是醒来到一个满是红色的仪表板, 把你的AI代理工具 自动给您的 Dagster 实例。 为自动 DataOps 工作流程设计的跨6个类别的27个工具: 动作是选入的:设置 DAGSTER READ ONLY=假以启用写入操作. - Python 3.12+ - uv(建议)或 pip - 运行中的 Dagster 实例(自主办或云) 在一个服务器中使用 DAGSTER ENVS 配置多个 Dagster 实例 。 每个工具然后接受一个可选的env参数,这样LLM就可以瞄准正确的实例.
概述（AI 翻译）
一个MCP服务器,它让AI代理 完全能见度和控制您的 Dagster 实例—— 就像一个用于您数据管道的 SRE.
与任何MCP客户端合作:克洛德代码,克洛德桌面,Cursor等.
Dagster正式开发并支持他们自己的Dagster Plus MCP服务器. 见这里的文件。
□为什么这个存在
数据管道在凌晨3点中断。 时间表静静地停止开火。 资产沉闷。 而不是醒来到一个满是红色的仪表板上, 给你的人工智能代理器 用于监视器,诊断, 并自主地修复你的 Dagster 病例。
它会做什么#
为自动 DataOps 工作流程设计的跨6个类别的27个工具:
· 类 · 工具 · · 经纪人能做什么? |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- **** 运行** 查找失败、诊断根源、检查日志和步骤时间QQ ** ** 资产** [[T5] [T6] [T7] [T8] [T9] 发现资产、预览行中选择、检查新鲜度、检测陈旧数据* *
本页由 AI 翻译，技术细节请以原文为准。
README
以下为原文 README 节选，完整内容请查看原文。
Dagster MCP
An MCP server that gives AI agents full visibility and control over your Dagster instance — like an SRE for your data pipelines.
Works with any MCP client: Claude Code, Claude Desktop, Cursor, and more.
Dagster officially develops and supports their own Dagster Plus MCP server. See the documentation here.
Why this exists
Data pipelines break at 3 AM. Schedules silently stop firing. Assets go stale. Instead of waking up to a dashboard full of red, give your AI agent the tools to monitor, diagnose, and fix your Dagster instance autonomously.
Agent: Checking instance health...
get_instance_status() -> healthy: false, daemon "SCHEDULER" unhealthy
Agent: Scheduler daemon is down. Let me check recent failures...
get_runs(statuses=["FAILURE"], limit=5) -> 3 failed runs in the last hour
Agent: Diagnosing the most recent failure...
get_run_failure_summary("run_abc123") ->
failed_steps: ["transform_orders"]
root_cause: "NullPointerError: column 'price' is null"
suggestions: ["Single step failed — consider re-running from failure"]
Agent: Re-launching the failed job...
launch_job("etl_pipeline", "my_project") -> run_id: "run_def456", status: STARTED
Agent: One sensor keeps firing on bad data. Checking its ticks...
get_tick_history("orders_sensor", "SENSOR") -> 12 consecutive FAILURE ticks
Agent: Stopping it until the upstream fix lands...
stop_sensor("orders_sensor") -> status: STOPPED
What it does
27 tools across 6 categories, designed for autonomous DataOps workflows:
| Category | Tools | What an agent can do |
|---|---|---|
| Runs | get_runs get_run_status get_run_logs get_run_stats get_run_failure_summary |
Find failures, diagnose root causes, inspect logs and step timing |
| Assets | search_assets resolve_asset_selection get_asset_details get_recent_materializations get_asset_health |
Discover assets, preview lineage selections, check freshness, detect stale data |
| Jobs | list_jobs |
Inventory jobs across code locations, with optional repository/location filters |
安装
uvx dagster-mcp配置
{
"mcpServers": {
"dagster": {
"command": "uvx",
"args": ["dagster-mcp"],
"env": {
"DAGSTER_URL": "http://localhost:3000"
}
}
}
}