Dagster Mcp

Dagster MCP

一个 MCP 服务，让 AI Agent 全面查看和控制你的 Dagster 实例——就像是数据管道的 SRE。

兼容任何 MCP 客户端：Claude Code、Claude Desktop、Cursor 等。

Dagster 官方开发并支持自己的 Dagster Plus MCP 服务。请参阅此处的文档。

为什么要开发它

数据管道会在凌晨 3 点发生故障。计划任务可能悄无声息地停止触发。资产会变得陈旧。与其醒来后面对满屏红色的仪表盘，不如为 AI Agent 提供工具，让它自主监控、诊断和修复你的 Dagster 实例。

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

功能

提供 6 类共 27 个工具，专为自主 DataOps 工作流设计：

操作默认关闭：设置 DAGSTER_READ_ONLY=false 以启用写入操作。

快速开始

前提条件

• Python 3.12+
• uv（推荐）或 pip
• 正在运行的 Dagster 实例（可自托管或使用 Cloud）

安装

此软件包已发布到 PyPI。

选项 A — 使用 uvx 直接运行（无需安装）：

uvx dagster-mcp

选项 B — 使用 pip 安装：

pip install dagster-mcp

选项 C — 克隆并运行：

git clone https://github.com/fabdendev/dagster-mcp.git
cd dagster-mcp
uv sync

配置

单一环境

自托管：

export DAGSTER_URL=http://localhost:3000

Dagster Cloud：

export DAGSTER_URL=https://myorg.dagster.cloud/prod
export DAGSTER_API_TOKEN=your-dagster-cloud-user-token

自定义身份验证／代理请求头：

export DAGSTER_EXTRA_HEADERS='{"Authorization":"Bearer your-token","X-My-Header":"value"}'

多个环境

使用 DAGSTER_ENVS 在一台服务器中配置多个 Dagster 实例。之后，每个工具都会接受可选的 env 参数，以便 LLM 定位到正确的实例。

export DAGSTER_ENVS='{
  "prod": {"url": "https://myorg.dagster.cloud/prod", "token": "prod-token"},
  "staging": {"url": "https://myorg.dagster.cloud/staging", "token": "stg-token"},
  "dev": {"url": "http://localhost:3000"}
}'
export DAGSTER_DEFAULT_ENV=prod

设置 DAGSTER_ENVS 后，DAGSTER_URL / DAGSTER_API_TOKEN / DAGSTER_EXTRA_HEADERS 将被忽略。如果只定义了一个环境，即使未设置 DAGSTER_DEFAULT_ENV，也会自动使用该环境。

添加到 MCP 客户端

{
  "mcpServers": {
    "dagster": {
      "command": "uvx",
      "args": ["dagster-mcp"],
      "env": {
        "DAGSTER_URL": "http://localhost:3000"
      }
    }
  }
}

{
  "mcpServers": {
    "dagster": {
      "command": "uvx",
      "args": ["dagster-mcp"],
      "env": {
        "DAGSTER_ENVS": "{\"prod\":{\"url\":\"https://myorg.dagster.cloud/prod\",\"token\":\"prod-token\"},\"dev\":{\"url\":\"http://localhost:3000\"}}",
        "DAGSTER_DEFAULT_ENV": "prod"
      }
    }
  }
}

{
  "mcpServers": {
    "dagster": {
      "command": "uvx",
      "args": ["dagster-mcp"],
      "env": {
        "DAGSTER_URL": "http://localhost:3000"
      }
    }
  }
}

{
  "mcpServers": {
    "dagster": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/dagster-mcp", "dagster-mcp"],
      "env": {
        "DAGSTER_URL": "http://localhost:3000"
      }
    }
  }
}

工具参考

Runs

Assets

两步资产工作流

首先解析并检查选择表达式：

resolve_asset_selection(
  asset_selection="group:analytics and (kind:dbt or key:*benchmark)"
)

→ {
    "asset_keys": [
      "warehouse/analytics/orders",
      "warehouse/analytics/reranker_benchmark"
    ],
    "assets": [...]
  }

对于具体的非分区资产，将返回的键以及所需的启动配置和标签传递给 materialize_assets：

materialize_assets(
  asset_keys=[
    "warehouse/analytics/orders",
    "warehouse/analytics/reranker_benchmark"
  ],
  run_config={"ops": {"benchmark": {"config": {"limit": 1000}}}},
  tags={"triggered_by": "agent"}
)

对于分区资产，则将相同的已解析键传递给 backfill_assets：

backfill_assets(
  asset_keys=["warehouse/analytics/daily_orders"],
  partition_start="2026-07-01",
  partition_end="2026-07-07",
  run_config={"resources": {"warehouse": {"config": {"pool": "benchmark"}}}}
)

只读和读写模式均可使用 resolve_asset_selection。 它会返回外部、可观察、不可执行和分区匹配项，以便调用方检查完整结果。写入工具会在执行前重新获取当前资产定义。resolve_asset_selection 和 materialize_assets 要求 Dagster 1.9+。

作业、计划与传感器

list_jobs 的筛选条件相互独立，组合使用时采用 AND 语义。例如，list_jobs(repository_name="example_repository") 会跨代码位置查找该存储库，而 list_jobs(repository_name="example_repository", location_name="example_location") 则指定一个存储库与代码位置组合。只使用一个筛选条件的调用会先发送轻量级的存储库发现请求，再批量请求作业信息；这并非免费：它会多产生一次与 Dagster 的往返请求，以便在只有部分存储库匹配时缩小负载。每次调用的请求数：无筛选条件时为 1，同时使用两个筛选条件时为 1，仅使用一个筛选条件时为 2。

如果与筛选条件相关的代码位置加载失败或仍在加载，list_jobs 会抛出包含该位置名称的 RuntimeError，而不是悄然返回空结果——筛选结果为空，意味着 Dagster 实际能够加载的代码位置中没有匹配项。筛选结果非空时，仍可能不完整：如果有些存储库匹配，list_jobs 就会返回这些存储库，而不会抛出异常，即使另一个代码位置无法搜索也是如此。此检查会随上述相同请求一并进行（workspaceOrError，list_code_locations 本来就会查询它），因此不会增加额外的往返请求。不带筛选条件的调用（无筛选条件的 list_jobs()）保持不变：它会省略此检查，并继续返回当前已加载的所有内容，因为 Agent 不会像误读空筛选结果那样，把冗长的清单误读为“什么都不存在”——可使用 list_code_locations 或 get_instance_status 直接检查加载状态。

实例与代码位置

写入操作

调度/传感器名称仅在仓库内唯一。如果多个代码位置存在同名对象，启动/停止工具会拒绝操作并列出候选项；传入 repository_name / location_name 可消除歧义。调用成功时会回显解析出的 repository 和 location。 | terminate_run | 停止卡住或失控的运行 | | reload_code_location | 部署后重新加载代码位置 |

写入工具要求 DAGSTER_READ_ONLY=false（默认为 true）。

与 Dagster 官方 AI 工具的区别

Dagster 构建并支持自己的 AI 工具。如果你是 Dagster+ 客户，请优先使用官方工具——它们由 Dagster 一方提供，无需本地运行时，并且可以访问 Dagster+ 平台对象（警报、Issues、Insights），而本项目无法访问这些对象。本项目主要面向运行开源/可自托管 Dagster 的用户；官方服务器文档中的设置说明未涵盖这类用户。

“官方”列中的所有内容均来自 Dagster 自己的文档 https://docs.dagster.io/guides/labs/dagster-mcp（检索日期：2026-07-30），或来自 Dagster 在 #21 中的直接说明。Dagster 的文档并未声明不支持开源或可自托管的 Dagster；文档说明的是由 Dagster 托管的 URL、必需的 Dagster-Cloud-Organization 标头，以及 Dagster+ 用户令牌。其功能矩阵只是当前情况的快照，Dagster 也预计会继续扩展功能，因此请将下方的差异视为当前情况，而非永久不变。

如果你运行的是开源版或可自托管的 Dagster，或者需要通过 Agent 执行资产级物化、计划/传感器控制、回填，或代码位置和守护进程操作，请使用本项目。

对于 Dagster+ 用户，两者是互补而非竞争关系——你完全可以同时运行它们。本项目与 Dagster Labs 无关联，也未获得其认可。

兼容性

监控工具和现有操作工具已通过 Dagster 1.6+ 测试。 resolve_asset_selection 和 materialize_assets 需要 Dagster 1.9+，并且会在查询或启动前验证所需的 GraphQL 功能。RunsFilter 字段名称（jobName 与 pipelineName）会通过架构自省自动检测。配置的资产回填也会检测 LaunchBackfillParams.runConfigData 是否存在；如果较旧的架构未公开该字段，则会返回明确的兼容性错误。计划和传感器的启动/停止适用于 Dagster 1.6+：停止 mutation 使用 originId/selectorId 参数对，其值来自 InstigationState.id/selectorId；现代 Dagster 会将其重新解析为复合 ID，而较旧版本可直接接受，因此无需按版本分支处理。mutation 通过 ... on Error 接口片段选择错误消息，而非使用按类型划分的片段，因为 ScheduleNotFoundError 直到 Dagster 1.9 才加入 ScheduleMutationResult；若在 1.6-1.8 中展开该片段，会导致文档验证失败。

开发

uv sync --extra dev
uv run ruff check dagster_mcp/    # lint
uv run pytest                     # run the test suite
uv run python -m dagster_mcp      # start server locally

许可证

MIT