Claude CodeでDagsterのアセットを定義する — テストしやすい宣言的パイプラインの書き方

Dagsterのアセットとは

Dagsterのアセットとは、データパイプラインが生成する成果物(テーブルやファイルなど)をPythonの関数として宣言的に定義したものです。実行順序のタスクを直接書くのではなく、「このデータはどう作られるか」を関数単位で書くと、Dagsterが依存関係のグラフを組み立てます。

@dg.asset デコレータを関数に付けると1つのアセットになり、deps 引数で上流のアセットを指定すれば依存関係も明示できます。最小構成は次のとおりです。

import dagster as dg
 
@dg.asset
def hello(context: dg.AssetExecutionContext):
    context.log.info("Hello!")
 
@dg.asset(deps=[hello])
def world(context: dg.AssetExecutionContext):
    context.log.info("World!")

Claude Codeに書かせる側から見ると、アセット定義そのものは通常のPythonコードです。Dagster特有なのは、プロジェクトの作り方、dg CLIの使い方、テストの書き方の3点で、この記事はその順に進みます。

Claude CodeでDagsterプロジェクトを立ち上げる手順

新規プロジェクトの作成からアセットのスキャフォールド(雛形生成)までは、Dagster公式のCLIをClaude Codeに実行させれば数コマンドで終わります。前提はPython 3.10以上で、パッケージ管理にはuvかpipを使います。以下はuvの手順です。

手順1: プロジェクトを作成する

uvx create-dagster@latest project dagster-quickstart
cd dagster-quickstart
source .venv/bin/activate
uv add pandas

create-dagster CLIはpyproject.tomlとsrc以下のパッケージ構成を生成し、プロンプトでyと答えるとuv syncまで実行します。pipの場合はcreate-dagster projectのあと、仮想環境を手動で作り、pip install --editable . --group devでプロジェクトを編集可能な状態にします。--groupはpip 25.1以降のオプションです。

手順2: アセットファイルをスキャフォールドする

dg scaffold defs dagster.asset assets.py

生成物は、実行したディレクトリに関わらず<project-name>/src/<project_name>/defs/配下に置かれます。生成直後はコメントアウトされた雛形なので、ここからClaude Codeに実装を書かせます。

手順3: 定義を確認して起動する

dg list defs
dg dev

dg list defsはプロジェクト内のアセットや依存関係を一覧します。Claude Codeが書いたコードがDagsterに認識されているかを、実行せずに確かめられます。dg devはローカルのDagsterインスタンス(UI付き)を起動するコマンドで、プロジェクトのディレクトリで実行すると、そのプロジェクトだけが読み込まれます。

どのデコレータでアセットを書かせるか

Dagsterのアセットデコレータは4種類あります。Claude Codeに実装を依頼する前に、どれを使うかを決めて伝えておくと、生成されるコードの構造が揃います。

@dg.multi_assetはAssetSpecのリストで出力を宣言し、MaterializeResultをyieldして各アセットの結果を返します。関数名のmy_multi_assetはアセットではなく、asset_oneとasset_twoのどちらを実体化しても、この関数が実行されます。

import dagster as dg
 
@dg.multi_asset(specs=[dg.AssetSpec("asset_one"), dg.AssetSpec("asset_two")])
def my_multi_asset():
    yield dg.MaterializeResult(asset_key="asset_one", metadata={"num_rows": 10})
    yield dg.MaterializeResult(asset_key="asset_two", metadata={"num_rows": 24})

アセットにはcode_versionを付けられます。コードが変わったのに再実体化していないアセットをDagsterが追跡でき、同じ計算の繰り返しも避けられます。UIは、最新の実体化時とcode_versionが異なるアセットを示します。アセット名が関数名から決まる点は、MCP Python SDKのデコレータでツールを定義する実装パターンで関数名がツール名になる点と同じ発想です。

アセットのキーとメタデータを設計する

アセットにはグループ名、所有者、複数階層のキーを付けられます。プロジェクト全体の実装を任せるなら、先に決めておくと後から探しやすくなります。

group_nameは関連するアセットをまとめる名前です。ownersには担当者のメールアドレスやteam:roofのようなチーム名を並べます。

import dagster as dg
 
@dg.asset(deps=[daily_sales], group_name="sales")
def weekly_sales(context: dg.AssetExecutionContext) -> None:
    context.log.info("Loading weekly sales")
 
@dg.asset(deps=[weekly_sales], owners=["team:roof"])
def weekly_sales_report(context: dg.AssetExecutionContext) -> None:
    context.log.info("Building the report")

ファイルシステムのように階層を持つアセットは、key_prefixに文字列のリストを渡して複数階層のキーにします。下流のアセットはAssetInに同じkey_prefixを指定して上流を参照します。

import dagster as dg
 
@dg.asset(key_prefix=["one", "two", "three"])
def upstream_asset():
    return [1, 2, 3]
 
@dg.asset(ins={"upstream_asset": dg.AssetIn(key_prefix=["one", "two", "three"])})
def downstream_asset(upstream_asset):
    return upstream_asset + [4]

グループ名の単位やキーの階層の切り方は、CLAUDE.mdに書いておくと、生成のたびに表記を指示し直さずに済みます。

アセットの単体テストはどう書くか

Dagsterのアセットは普通のPython関数なので、Dagsterのランタイムを起動せずに直接呼び出せます。Claude Codeが書いたアセットも、その場でテストにできます。どの書き方になるかは、アセットの引数で決まります。

上流に依存するアセットは、その依存先の値を直接渡して呼び出せます。configを受け取るアセットは、Configオブジェクトを作って渡します。引数が複数あるときは、キーワード引数で渡す書き方が推奨されています。個々のアセットをテストするほうが、ジョブ全体をテストするより一般的に勧められています。

実行は、testsディレクトリを含むプロジェクトの最上位ディレクトリで行います。

pytest tests

Databricksのジョブを直接呼ぶアセットのように、業務ロジックの大半が外部システムにある場合、単体テストは推奨されていません。

実体化したデータの中身を検証するには、単体テストとは別にアセットチェックを使います。単体テストは関数のロジックを確かめ、アセットチェックは実行後のデータが期待どおりかを確かめる、という役割の違いがあります。

dagster-expertスキルとDagster+ MCPサーバーを入れる

Dagsterは、Claude Code向けのdagsterプラグインをGitHubのdagster-io/skillsで配布しています。プラグインには、dagster-expertスキルとDagster+ MCPサーバーの両方が入ります。スキルにはdg CLIの使い方やアセットのパターンが書かれていて、スラッシュコマンドとして呼び出せます。

/plugin marketplace add dagster-io/skills
/plugin install dagster@dagster
/dagster-expert create a new Dagster project called my-pipeline

プラグインを使わずスキルだけ入れる場合は、npx skills add dagster-io/skills#release-stableも使えます。

導入後は/pluginのInstalledタブでdagsterが有効になっているかを確かめます。2026年9月14日から、旧プラグインのdagster-expertは非推奨で、指針を持たないスタブに置き換わっています。旧プラグインを入れていた場合は、アンインストールしてからdagsterプラグインを入れ直します。スキル自体は変わらず、/dagster-expertで呼び出します。

MCPサーバーで操作できる対象は、公式の表で次のように分かれます。

Dagster+のIssuesは、取得・作成・更新・削除の4つとも可能です。Insightsのメトリクスは、Runs、Assets、Deploymentsが対象です。

プラグインを使わず単体で接続するなら、次のコマンドです。

claude mcp add --transport http dagster-plus https://mcp.agent.dagster.cloud/mcp

接続後は/mcpでdagster-plusを選び、Authenticateでブラウザ認証を済ませます。OAuthで認証すると、MCPサーバーの権限はログインしたユーザーの権限になります。別の権限にしたいときは、サービスユーザーを作り、--headerでトークンと組織名を渡します。EUリージョンの組織は、URLをhttps://mcp.agent.eu.dagster.cloud/mcpに変えます。プラグイン経由なら、環境変数DAGSTER_CLOUD_MCP_URLで切り替えます。

導入のスコープは3つのコマンドで既定が違う

プラグインを入れる前に、--helpで既定のスコープを確かめておくと、チームへの配り方を決めやすくなります。Claude Code v2.1.289で実行した出力の抜粋です。

$ claude --version
2.1.289 (Claude Code)
 
$ claude plugin marketplace add --help
  --scope <scope>      Where to declare the marketplace: user (default),
                       project, or local
 
$ claude plugin install --help
  -s, --scope <scope>  Installation scope: user, project, or local
                       (default: "user")
 
$ claude mcp add --help
  -s, --scope <scope>  Configuration scope (local, user, or project)
                       (default: "local")

マーケットプレイスの追加とプラグインの導入は、既定がuserです。userはプロジェクトを問わず自分の環境に入れる指定です。一方、claude mcp addの既定はlocalで、扱いが異なります。チームで同じ構成にしたいときは、3つとも--scope projectを明示します。

導入後は、claude plugin details dagsterでコンポーネントの一覧と、見積もりのトークンコストを確かめられます。MCPサーバーが公開するツールが増えるほどコンテキストを消費する点は、MCPのツール定義はなぜコンテキストを圧迫するのかで扱った制約と同じ理屈です。

よくあるつまずき

• 仮想環境を有効化し忘れる: dg CLIはcreate-dagsterで作ったプロジェクトの仮想環境を有効化して使います。source .venv/bin/activateを先に実行します。
• multi_assetの依存を単純にテストしようとする: 上流に依存する@multi_assetはinput_valuesが使えず、モックのIOマネージャーが要ります。
• 旧プラグインのdagster-expertを入れてしまう: 指針を持たないスタブに置き換わっています。dagster@dagsterを入れ直します。

まとめ

Claude CodeでDagsterを書くときに差が出るのは、コード生成より、dg CLIの実行とアセットの引数に合わせたテスト設計です。ローカルでdg scaffold defs、dg list defs、直接呼び出しのテストを一往復させ、デプロイ済みのDagster+を会話から操作したくなった段階でMCPサーバーを足す順番だと、導入の負担が小さく済みます。