Dagster Mcp

Dagster MCP

AI エージェントに Dagster インスタンスの完全な可視性と制御機能を提供する MCP サーバーです。データパイプラインの SRE のように機能します。

あらゆる MCP クライアントで利用できます：Claude Code、Claude Desktop、Cursor など。

Dagster は、独自の Dagster Plus MCP サーバーを公式に開発、サポートしています。ドキュメントはこちらをご覧ください。

このプロジェクトの目的

データパイプラインは午前3時に停止します。スケジュールがいつの間にか実行されなくなります。アセットが古くなります。赤く染まったダッシュボードを見て目覚める代わりに、AI エージェントにツールを与えて、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

できること

自律的な DataOps ワークフロー向けに設計された、6カテゴリにわたる27個のツールを備えています：

アクションはオプトイン方式です。書き込み操作を有効にするには、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 を使用して、1 つのサーバーで複数の 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 は無視されます。環境が 1 つだけ定義されている場合は、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"
      }
    }
  }
}

ツールリファレンス

実行

アセット

2 段階のアセットワークフロー

まず、選択式を解決して内容を確認します。

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") は、1 つのリポジトリとコードロケーションの組み合わせを対象にします。フィルターを 1 つだけ指定した場合は、軽量なリポジトリ検出リクエストに続いて、ジョブを一括取得するリクエストが送信されます。追加のリクエストが不要になるわけではありません。Dagster への往復が 1 回増える代わりに、一致するリポジトリの一部だけを対象とする場合にペイロードを小さくできます。呼び出しごとのリクエスト数は、フィルターなしの場合は 1、両方のフィルターを指定した場合は 1、フィルターを 1 つだけ指定した場合は 2 です。

フィルターに該当するコードロケーションの読み込みに失敗した場合、または読み込み中の場合、list_jobs は空の結果を黙って返すのではなく、そのロケーション名を示す RuntimeError を発生させます。フィルター適用後の結果が空であれば、Dagster が実際に読み込めたコードロケーション内に一致するものがないことを意味します。フィルター適用後の結果が空でなくても、不完全な場合があります。一部のリポジトリが一致すれば、別のコードロケーションを検索できなかった場合でも、list_jobs はエラーを発生させずにそれらを返します。このチェックは上記と同じリクエスト（list_code_locations がすでに照会している workspaceOrError）に含まれるため、追加の往復は発生しません。フィルターなしの呼び出し（フィルターを指定しない list_jobs()）は変更されていません。このチェックを省略し、読み込み済みの情報をそのまま返します。エージェントが長い一覧を「何も存在しない」と誤解することはありませんが、フィルター適用後の空の結果ではそのような誤解が起こり得るためです。読み込み状態を直接確認するには、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+ プラットフォームのオブジェクト（アラート、Issues、Insights）にもアクセスできます。このプロジェクトは主に、公式サーバーのドキュメントに記載されたセットアップではカバーされていない、オープンソース版／セルフホストの Dagster を実行する方向けです。

「公式」列の内容はすべて、Dagster の公式ドキュメント https://docs.dagster.io/guides/labs/dagster-mcp（2026-07-30 取得）または #21 で Dagster から直接提供された情報に基づいています。Dagster のドキュメントには、オープンソース版またはセルフホストの Dagster がサポート対象外であるとはどこにも記載されていません。ドキュメントに記載されているのは、Dagster がホストする URL、必須の Dagster-Cloud-Organization ヘッダー、および Dagster+ ユーザートークンです。機能マトリックスは現時点の状況を示すもので、Dagster は今後も機能を追加していく予定です。そのため、以下の不足点は恒久的なものではなく、現時点のものとしてご覧ください。

次の条件に当てはまる場合は、このプロジェクトを使用してください。 オープンソースまたはセルフホスト可能な Dagster を運用している場合、あるいはエージェントからアセットレベルのマテリアライゼーション、スケジュール／センサーの制御、バックフィル、コードロケーションやデーモンの操作を行う必要がある場合です。

Dagster+ のユーザーにとって、両者は競合するものではなく、補完し合うものです。両方を実行しても問題ありません。このプロジェクトは Dagster Labs と提携しておらず、同社の推奨を受けたものでもありません。

互換性

監視ツールと既存のアクションツールは Dagster 1.6+ でテストされています。 resolve_asset_selection と materialize_assets には Dagster 1.9+ が必要であり、 クエリまたは起動の前に必要な GraphQL 機能を確認します。 RunsFilter のフィールド名（jobName と pipelineName）は、スキーマの イントロスペクションによって自動検出されます。設定済みのアセットバックフィルも LaunchBackfillParams.runConfigData の有無を機能検出し、古いスキーマでこのフィールドが 公開されていない場合は、明確な互換性エラーを返します。スケジュールとセンサーの開始／停止は Dagster 1.6+ で動作します。停止ミューテーションは、InstigationState.id／selectorId から 取得した originId／selectorId の引数ペアを使用して送信されます。最新の Dagster では これが複合 ID として再解析され、古いバージョンではそのまま受け入れられるため、 バージョンごとの分岐は不要です。ミューテーションでは、型ごとのフラグメントではなく ... on Error インターフェースフラグメントを使ってエラーメッセージを選択します。 ScheduleNotFoundError が ScheduleMutationResult に加わったのは Dagster 1.9 であり、 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