アーキテクチャ

エージェントオーケストレーション型データパイプライン：Dagster + dbt + MCPスタックの構築

著者：IdeaBosque最終更新：2026年8月28日

ポイント

Fivetranとdbt Labsは2026年6月1日に合併を完了し（合計ARR約$600M、10万以上のデータチーム）、Agents Schemaを提供した——これはウェアハウススキーマをエージェント向けのガバナンスされた共有コンテキスト層に変えるオープン標準である。
Databricksは、Neonユニット上のデータベースの80%以上が人間ではなくAIエージェントによってプロビジョニングされていると報告している——データスタックの主な利用者は、すでにアナリストからエージェントへと移行した。
新しいパイプラインを定義する3つのエージェント型パターン——エージェントがパイプラインコードを記述・雛形化する、パイプラインが修正を提案・適用して自己修復する、エージェントがチャットで実行失敗をトリアージする——のいずれも、レンダリングされたダッシュボードではなく、機械可読なリネージを必要とする。
スタックは3つのガバナンスされた層で構成される——アセットリネージを担うDagster、テスト済み変換と共有コンテキストを担うdbt、ポリシーでスコープされたアクセスを担うMCPモジュール——これにより、エージェントは人間と同じ統制の下でパイプラインの構造を読み取る。

データスタックは人間のアナリスト向けに構築されてきた：夜間にパイプラインを実行し、朝にダッシュボードを読み、数値がおかしければチケットを起票する。AIエージェントはデータの消費の仕方が異なる。合併後のFivetran + dbt Labsが述べているように、エージェントは「継続的に、並行して、マシン速度で動作する」——そして彼らが必要とするのはパイプラインの構造（リネージ、テスト、定義）であり、単なる出力ではない。わずか1四半期のうちに、最大手のデータ移動・変換ベンダーはその事実を軸に自らを再構築した。Fivetran + dbt合併はAgents Schemaを提供し、dbt Fusionエンジンをdbt Core v2.0としてオープンソース化した。またDatabricksはElectricを買収し、各エージェントに使い捨ての専用Postgresを与えた。

このガイドでは、B2B調達の文脈でこのパターンを構築する：サプライヤーカタログ、価格、在庫を取り込み、それらをRFQエージェントに公開するパイプラインだ。取り上げるのは3つの層——アセット中心のオーケストレーションを担うDagster、ガバナンスされた変換を担うdbt、スコープされたエージェントアクセスを担うMCPモジュール——と、パイプラインを自己維持させる3つのエージェント型パターンだ。読み終える頃には、各層が何を担っているか、なぜアセットリネージが要となる選択なのか、そして人間がどこでループにとどまるのかが分かるはずだ。

アセットセントリック・オーケストレーションが基盤である理由

エージェント向けオーケストレーションの失敗の多くは、誤ったメンタルモデルから始まる。タスク中心のスケジューラー（従来のcron＋DAG設計）が答えるのは「このジョブは実行されたか？」だ。「なぜAcmeの価格が古いのか？」と尋ねるエージェントには別の答えが必要になる。「どのデータアセットが古くなっており、何に依存し、何がそれを供給しているのか？」だ。これはアセットに関する問いであり、だからこそDagsterのアセットセントリックモデルが単なる好みではなく基盤なのだ。

Dagsterでは、生成するもの——supplier_catalog、normalized_prices、availability_snapshot——とその依存関係を宣言する。オーケストレーターはそれによって完全なリネージグラフを把握する。DagsterのDeclarative Automationは、固定スケジュールではなく上流の変更に応じてアセットを更新できるようにするため、「古い」がシステムが推論できるプロパティになる。このリネージグラフこそ、エージェントが推測なしに不正な数値を発生源までたどるために必要なものだ。

import dagster as dg

@dg.asset(group_name="procurement")
def supplier_catalog(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    rows = fetch_supplier_feed()  # NetSuite, EDI, CSV drop, etc.
    write_bronze("supplier_catalog", rows)
    return dg.MaterializeResult(metadata={"row_count": len(rows)})

@dg.asset(deps=[supplier_catalog], group_name="procurement",
          automation_condition=dg.AutomationCondition.eager())
def normalized_prices() -> None:
    # dbt owns the transformation logic; Dagster owns the lineage + trigger
    run_dbt(select="normalized_prices")

depsとautomation_conditionこそが要点だ。エージェント（および下記の自己修復ループ）は、このグラフをデータとして読み取れる。Airflowも別方向から同じ結論に至った——Airflow 3.2はCommon AI Providerとアセット対応スケジューリングを追加した——そして業界再編は現実のものだ。Prefectは2026年7月にDagsterを買収した。どのオーケストレーターに標準化するとしても、要件は同じだ。不透明なタスクではなく、宣言されたリネージを持つアセットであることだ。

dbtが変換とガバナンスされたコンテキストを担う理由

Dagsterは作業をトリガーしリネージを追跡するが、そこにビジネスロジックを持たせるべきではない。それはdbtの役割であり、そこではすべての変換がテスト、ドキュメント、セマンティック定義を伴うバージョン管理されたSQLモデルとなる。エージェントにとってこれは付加価値ではなく、信頼の境界そのものだ。dbt自身の立場は、変換層こそがエージェント型パイプラインを信頼できるものにするというものだ。定義されておらずテストもされていないテーブルに対してSQLを書くエージェントは、混乱をより速く自動化するだけである。

合併後の追加機能で最も重要なのはAgents Schemaだ。これは、メトリック定義、セマンティックモデル、dbtリネージ、業務ドキュメントをプレーンなSQLテーブルとして保存する専用のウェアハウススキーマである。すべてのエージェントが「手持ち在庫」の意味を個別に再導出する代わりに、その定義はエージェントが参照する単一のガバナンスされた顧客保有の場所に存在する。これはガバナンスされたコネクタモジュールのデータ側の対になるものであり、ドリフトするエージェントごとのコピーではなく、単一のポリシーでスコープされた共有コンテキストのソースだ。

-- models/marts/availability_snapshot.sql
select
    sku,
    warehouse_id,
    on_hand - allocated as available_qty,   -- the governed definition
    updated_at
from {{ ref('normalized_inventory') }}

-- schema.yml: the test that gates the agent's trust
-- - name: available_qty
--   tests: [not_null, {dbt_utils.accepted_range: {min_value: 0}}]

テストが失敗するということは、エージェントがその行を根拠に見積もりを出すべきではないというシグナルだ。すべてのモデルに機械可読な合否判定があるというこの一点こそが、次の2つのパターンを人間が逐一監視せずに動かせるようにしている。

エージェントの接続先：データベースログインではなくMCPモジュール

エージェントは生のウェアハウス認証情報を持つべきではない。エージェントが呼び出すべきは、少数の型付きツール——get_availability(sku, warehouse)、get_tier_price(sku, customer_tier)、list_substitutes(sku)——を公開するガバナンスされたMCPモジュールであり、それぞれがテスト済みのdbtモデルにマッピングされ、ポリシースコープ、レート制限、監査ログを備えている。これはERPやコマースコネクタで使われているのと同じモジュールパターンを、パイプライン自身の出力に適用したものだ。これにより爆発半径は小さく保たれる。エージェントはavailability_snapshotを読み取れるが、任意のSQLを実行することはできず、すべての呼び出しがログに記録される。

その境界は、エージェントごとの状態が収まる場所でもある。DatabricksによるElectricの買収——エージェントサンドボックス内のWASM Postgres（PGlite）を中央のガバナンスされた状態と同期させるもの——が存在するのは、エージェントが作業用コンテキストのために「数千もの小さく使い捨て可能なデータベース」を必要とするからであり、それらは永続的でガバナンスされたテーブルとは分離して保たれる。原則はこうだ。永続的で共有され、ガバナンスされたデータはMCPモジュールの背後に置き、動きの速い実行ごとのスクラッチコンテキストはエージェント自身のサンドボックスに置く。

このスタックが可能にする3つのエージェント型パターン

リネージ（Dagster）、テスト済み定義と共有コンテキスト（dbt + Agents Schema）、そしてスコープされたアクセス（MCP）が揃うと、3つのパターンが現実的になる。

エージェント型開発。 エージェントは既存のリネージグラフに対して新しいアセットと変換を雛形化する——dbtモデルを草案化し、スキーマテストを提案し、Dagsterアセットを配線する。DagsterはClaude CodeとCodex向けのdagster-io/skillsと、Compass Slackアシスタントを提供しており、Bruinは同じ目的のMCPサーバーを公開している。人間がレビューするのは空白のファイルではなく、プルリクエストだ。
自己修復するパイプライン。 スキーマテストが失敗したり上流のアセットが壊れたりすると、エージェントはリネージを読み取り、失敗しているモデルを特定し、修正を提案して、カナリア実行で適用するか、あるいはPRを提出する。Dagsterが依存グラフを把握し、dbtがどのテストが失敗したかを把握しているため、その修正は盲目的な再試行ではなく、リネージを意識したものになる。
エージェント型トラブルシューティング。 失敗が起きると、エージェントは実行ログとメタデータを読み取り、SlackやTeamsで想定される原因と修正案を返信する——これはDagster CompassとSnowflake Cortexが前提としているパターンだ。オンコール対応は、ダッシュボードを読むことから、エージェントの診断をレビューすることへとシフトする。

これらのいずれも人間を排除するものではない。適用されるすべての変更は、テストゲート、カナリア、あるいはレビューのいずれかを通過する——これはdbt Summit 2026の基調講演が、エージェントに本番データを触らせる代償として位置づけた規律と同じものだ。

一望できるスタック

実際の構築事例

NetSuiteを稼働し、2つの倉庫と3つのサプライヤーカタログを持つある販売代理店は、人間が手作業で在庫を確認しなくても見積もりを出せるRFQエージェントを求めていた。ボトルネックはモデルではなくパイプラインだった。私たちはカタログ、価格、在庫の各アセットを明示的なリネージとともにDagsterで宣言し、価格・在庫ロジックを「引き渡し可能在庫」の定義を固定したAgents Schemaを備えたテスト済みdbtモデルに移し、読み取り専用にスコープされたMCPモジュールを通じて3つの型付きツールを公開した。自己修復ループは今や、朝の見積もり実行前に壊れたサプライヤーフィードを検知し、修正付きのPRを提出する。エージェントの最初の応答が障害ページではなく診断であるため、パイプライン障害に対するオンコール対応時間は減少した。エージェントはテスト済みのデータに基づいて見積もりを出すか、拒否する——テストに失敗した行を根拠に見積もりを出すことは決してない。