DataGrout Conduit SDK

DataGrout Conduit

mTLS、OAuth 2.1、セマンティックディスカバリーに対応した本番環境対応のMCPクライアント

標準MCPクライアントをそのまま置き換えられます。インポートを1行変更するだけで、ほかのコードは一切変更せずに、エージェントにセマンティックディスカバリー、コスト追跡、mTLS ID 管理、OAuth 2.1 を追加できます。

利用可能な言語

言語	パッケージ	インストール
Python	datagrout-conduit	pip install datagrout-conduit==0.8.1
TypeScript	@datagrout/conduit	npm install @datagrout/[email protected]
Rust	datagrout-conduit	cargo add [email protected]
Elixir	datagrout_conduit	{:datagrout_conduit, "~> 0.8.1"}
Ruby	datagrout-conduit	gem install datagrout-conduit -v 0.8.1

DataGroutを初めてお使いですか？まずはこちらをご覧ください。

DataGroutアカウントやサーバーURLをまだお持ちでない場合、SDKが自動で登録できます。ブラウザーも登録フォームも不要です。

Python

from datagrout.conduit import ClientBuilder
from datagrout.conduit.onramp import OnrampOptions

# One call: registers your agent, provisions a server, bootstraps mTLS identity.
# On subsequent runs the identity is auto-discovered from ~/.conduit/ — no token needed.
client = await ClientBuilder().bootstrap_onramp(OnrampOptions(
    gateway="https://app.datagrout.ai",
    agent_name="my-agent",
    agent_type="claude-sonnet-4-6",
    intended_use="Summarise documents and extract entities.",
))
await client.connect()

TypeScript

import { ClientBuilder } from '@datagrout/conduit';

const client = await new ClientBuilder()
  .bootstrapOnramp({
    gateway: 'https://app.datagrout.ai',
    agentName: 'my-agent',
    agentType: 'claude-sonnet-4-6',
    intendedUse: 'Summarise documents and extract entities.',
  });

await client.connect();

Rust

use datagrout_conduit::{ClientBuilder, onramp::OnrampOptions};

let client = ClientBuilder::new()
    .bootstrap_onramp(OnrampOptions {
        gateway: "https://app.datagrout.ai".into(),
        agent_name: "my-agent".into(),
        agent_type: Some("claude-sonnet-4-6".into()),
        intended_use: Some("Summarise documents and extract entities.".into()),
        access_code: None,
    })
    .await?
    .build()?;

client.connect().await?;

Elixir

{:ok, client} = DatagroutConduit.Client.bootstrap_onramp(%{
  gateway: "https://app.datagrout.ai",
  agent_name: "my-agent",
  agent_type: "claude-sonnet-4-6",
  intended_use: "Summarise documents and extract entities."
})

Ruby

client = DatagroutConduit::Client.bootstrap_onramp(
  gateway: "https://app.datagrout.ai",
  agent_name: "my-agent",
  agent_type: "claude-sonnet-4-6",
  intended_use: "Summarise documents and extract entities."
)
client.connect

内部で行われる処理：

1. SDKがエージェントのメタデータを添えて、/onramp にPOSTリクエストを送信します。
2. DataGroutが有効期間の短いセッショントークンを返します。
3. SDKが /onramp/complete でトークンを交換し、OAuth認証情報とプロビジョニングされたMCPサーバーURLを取得します。
4. ECDSA P-256キーペアがローカルで生成され、公開鍵がDataGrout CAに登録されます。署名済み証明書は ~/.conduit/ に保存されます。
5. 以降の実行では手順1～4がすべて省略され、ローカル証明書が直接提示されます。

秘密鍵がマシンの外に出ることはありません。認証情報とサーバーURLは保存されるため、証明書を紛失した場合も、次回以降の起動時に自動で復旧できます。

Invariant CLIを使いたい場合は、コードを書かずにターミナルから直接オンボーディングできます。

invariant onboard

サーバーURLを取得したら、以下のクイックスタートの例に示すように使用してください。

---

クイックスタート

Python

from datagrout.conduit import Client

async with Client("https://gateway.datagrout.ai/servers/{uuid}/mcp") as client:
    tools = await client.list_tools()
    result = await client.call_tool("salesforce@1/get_lead@1", {"id": "123"})

    results = await client.discover(query="find unpaid invoices", limit=10)
    session = await client.guide(goal="create invoice from lead")

TypeScript

import { Client } from '@datagrout/conduit';

const client = new Client('https://gateway.datagrout.ai/servers/{uuid}/mcp');
await client.connect();

const tools = await client.listTools();
const result = await client.callTool('salesforce@1/get_lead@1', { id: '123' });

await client.disconnect();

Rust

use datagrout_conduit::ClientBuilder;

let client = ClientBuilder::new()
    .url("https://gateway.datagrout.ai/servers/{uuid}/mcp")
    .auth_bearer("your-token")
    .build()?;

client.connect().await?;
let tools = client.list_tools().await?;

Elixir

{:ok, client} = DatagroutConduit.Client.start_link(
  url: "https://gateway.datagrout.ai/servers/{uuid}/mcp",
  auth: {:bearer, "your-token"}
)

{:ok, tools} = DatagroutConduit.Client.list_tools(client)
{:ok, result} = DatagroutConduit.Client.call_tool(client, "salesforce@1/get_lead@1", %{"id" => "123"})

Ruby

client = DatagroutConduit::Client.new(
  url: "https://gateway.datagrout.ai/servers/{uuid}/mcp",
  auth: { bearer: "your-token" }
)
client.connect

tools = client.list_tools
result = client.call_tool("salesforce@1/get_lead@1", { id: "123" })

---

認証

5つのSDKは、すべて同じ認証方式に対応しています。

ベアラートークン

client = Client("...", auth={"bearer": "your-token"})

OAuth 2.1（client_credentials）

client = Client("...", client_id="id", client_secret="secret")

SDKはJWTを取得、キャッシュし、自動更新します。

mTLS（相互TLS）

初回のブートストラップ後は、クライアント証明書で認証できるため、トークンは不要です。

# First run: bootstrap identity with a one-time token
client = await Client.bootstrap_identity(
    url="https://gateway.datagrout.ai/servers/{uuid}/mcp",
    auth_token="your-access-token",
    name="my-agent",
)

# All subsequent runs: mTLS auto-discovered from ~/.conduit/
# and presented directly to the same server's /mcp endpoint
client = Client("https://gateway.datagrout.ai/servers/{uuid}/mcp")

IDの自動検出では、次の順に検索します。

1. override_dir / identity_dir オプション（指定されている場合）
2. CONDUIT_MTLS_CERT + CONDUIT_MTLS_KEY 環境変数（インラインPEM）
3. CONDUIT_IDENTITY_DIR 環境変数（ディレクトリパス）
4. ~/.conduit/identity.pem + ~/.conduit/identity_key.pem
5. cwdを基準とした .conduit/

1台のマシンで複数のエージェントを使う場合は、identity_dir を指定して、各エージェントに独自の証明書ディレクトリを割り当ててください。

DataGrout CA

mTLS の ID は、DataGrout Certificate Authority（ca.datagrout.ai）によって署名された X.509 証明書です。DataGrout がサードパーティのプロバイダーに依存せず独自の CA を運用しているのは、エージェントの ID にはブラウザーの ID とは異なる要件があるためです。エージェントには、迅速なプログラムによる発行、自動ローテーション、人手による手続きを必要としないマシン間の信頼が求められます。

bootstrap_identity を呼び出すと、次の処理が行われます。

1. SDK がローカルで ECDSA P-256 鍵ペアを生成します。秘密鍵がマシンの外に出ることはありません。
2. 公開鍵が、ワンタイムアクセストークンまたは OAuth マシントークンとともに、対象サーバーの DG ID ブートストラップエンドポイントに送信されます。
3. DataGrout が、公開鍵を Substrate ID（例：CN=conduit-my-agent）に紐付け、その ID を特定の MCP サーバーに関連付けた証明書に署名します。
4. 署名済み証明書と CA チェーンが返され、ディスクに保存されます。

以降、すべてのリクエストでクライアント証明書が提示されます。サーバーは CA チェーンと登録済みのサーバー限定 ID に照らして証明書を検証します。トークンや、環境変数に保存するシークレットは不要で、手動で認証情報をローテーションする必要もありません。SDK が有効期限前に証明書を自動更新します。

CA の秘密鍵は HSM によって保護された AWS KMS キー（FIPS 140-2 Level 2）に保存されるため、署名鍵がメモリやディスク上に露出することはありません。CA 証明書は https://ca.datagrout.ai/ca.pem で公開されており、チェーンを独自に検証する必要があるクライアントは誰でも取得できます。

---

トランスポートモード

DataGrout は、3 種類すべてのトランスポートで同じインターフェイス、つまり同じツール、スキーマ、機能を提供します。mTLS、OAuth 2.1、ベアラートークン認証は、選択したトランスポートに関係なく同じように機能します。

トランスポート	プロトコル	用途
mcp（デフォルト）	MCP over Streamable HTTP / SSE	完全な MCP プロトコル。ストリーミング、通知に対応し、そのまま互換性があります
jsonrpc	HTTP POST over JSON-RPC 2.0	よりシンプルなプロトコル、ステートレス、1 リクエストにつき 1 レスポンス
websocket	WebSocket over JSON-RPC 2.0（datagrout-jsonrpc.v1）	双方向プッシュ。ポーリングせずにサーバー起点のイベントを購読できます

# MCP (default) — full protocol compliance, streaming support
client = Client(url)

# JSONRPC — same tools, same auth, simpler protocol
client = Client(url, transport="jsonrpc")

# WebSocket — bidirectional push subscriptions
client = Client("wss://gateway.datagrout.ai/servers/{uuid}/ws", transport="websocket")
sub = await client.subscribe("agents.my-agent-id.events")
async for event in sub:
    print(event.event, event.data)

WebSocket トランスポートは datagrout-jsonrpc.v1 サブプロトコルを使用します。単一の wss:// 接続がすべての同時リクエストで多重化され、同じソケット上で並行してツールを呼び出しても、サブスクリプションは維持されます。3 種類すべてのトランスポートで、すべてのツール機能を利用できます。違いはデータの受け渡し方式、つまり HTTP（プル）と WebSocket（プッシュ）です。

---

主な機能

インテリジェントインターフェイス

DataGrout サーバーへの接続時、デフォルトでインテリジェントインターフェイスが有効になり、すべてのツールが次の 2 つのツールに置き換わります。data-grout@1/discovery.discover@1 と data-grout@1/discovery.perform@1。エージェントが自然言語で実行したいことを説明すると、DataGrout がツールの解決、多重化、データ変換、グラフ作成などをバックグラウンドで処理します。無効にしてすべての生のツールを表示するには、use_intelligent_interface=False を指定します。

これはコンテキストウィンドウを大幅に最適化します。エージェントが数百、数千ものツールスキーマを推論する代わりに、2 つのツールだけを扱います。有効にすると、標準の tools/list のレスポンスは自動的に簡素化されたインターフェイスに置き換わります。内部では、perform が完全な機能セットをサポートします。複数のインテグレーションへのファンアウトを行う demux、データ変換を行う refract、グラフ生成、その他のサーバー側操作を、すべて単一の自然言語エントリーポイントから実行できます。

# With Intelligent Interface enabled, tools/list returns only 2 tools
tools = await client.list_tools()
# → [data-grout@1/discovery.discover@1, data-grout@1/discovery.perform@1]

# discover: semantic search for relevant tools by goal
results = await client.call_tool("data-grout@1/discovery.discover@1", {
    "goal": "find unpaid invoices",
    "limit": 10
})

# perform: execute a discovered tool by name with its args
result = await client.call_tool("data-grout@1/discovery.perform@1", {
    "tool": "quickbooks@1/get_invoices@1",
    "args": {"status": "unpaid"}
})

# perform also supports demux, refract, and chart as optional parameters:
result = await client.call_tool("data-grout@1/discovery.perform@1", {
    "tool": "quickbooks@1/get_invoices@1",
    "args": {"status": "unpaid"},
    "demux": True,          # fan-out across all connected servers with this tool
    "refract": "summarize by customer with totals",  # transform the output
    "chart": "bar chart of outstanding amounts by customer"  # generate a chart
})

セマンティックディスカバリー

インテリジェントインターフェイスを使用していない場合でも、discovery.discover を他のすべてのツールと併せてスタンドアロンのツールとして利用できます。正確なツール名を指定する必要はなく、目標との意味的な類似性に基づいて検索します。

results = await client.call_tool("data-grout@1/discovery.discover@1", {
    "goal": "find unpaid invoices",
    "limit": 5
})

コスト追跡

すべてのツール呼び出しは、クレジット使用量を含むレシートを返します。

from datagrout.conduit import extract_meta

result = await client.call_tool("salesforce@1/get_lead@1", {"id": "123"})
meta = extract_meta(result)
if meta:
    print(f"Credits: {meta.receipt.net_credits}")

ガイド付きワークフロー

複数ステップのワークフローを対話的に構築します。

session = await client.guide(goal="create invoice from lead")

Cognitive Trust Certificates

ワークフローに循環がなく、型安全で、ポリシーに準拠し、予算を尊重していることを証明する暗号学的証明です。CTC は、Substrate ID を発行するものと同じ DataGrout CA によって署名され、エージェントの ID からワークフローの検証までをつなぐ統一された信頼チェーンを構築します。

---

DataGrout ファーストパーティツール

標準の MCP tools/call に加え、すべての SDK で DataGrout サーバー API 全体を、型付きの名前空間メソッドとして利用できます。ドメイン固有のツールは、次の 6 つの名前空間に整理されています。

名前空間	用途
prism	データ変換、グラフ作成、レンダリング、エクスポート、型ブリッジ
logic	Prolog ロジックセルを基盤とする永続的なエージェントメモリ
warden	セーフティゲート、意図の検証、複数モデルによる合意形成
deliverables	成果物の登録、一覧表示、取得
ephemerals	キャッシュ管理 — キャッシュされた結果の一覧表示と検査
flow	ワークフローのオーケストレーション、ルーティング、ヒューマン・イン・ザ・ループ、実行履歴

コアとなる検出メソッド（discover、perform、guide、plan、dg）は引き続きクライアントに直接用意されています。

検出と計画

results = await client.discover(query="find unpaid invoices", limit=5)
plan = await client.plan(goal="onboard a new enterprise customer")
result = await client.perform("salesforce@1/get_lead@1", args={"id": "123"})
session = await client.guide(goal="create invoice from opportunity")

Prism：データ変換と可視化

transformed = await client.prism.refract(goal="group invoices by customer", payload=raw_data)

chart = await client.prism.chart(goal="bar chart of outstanding amounts", payload=invoice_data,
                                 chart_type="bar", x_label="Customer", y_label="Amount (USD)")

focused = await client.prism.focus(data=payload, source_type="invoice_list",
                                   target_type="crm_opportunity_list")

rendered = await client.prism.render(goal="executive summary", payload=data)
exported = await client.prism.export(content=data, format="csv")

Logic Cell（エージェントメモリ）

await client.logic.remember(statement="Customer Acme Corp has net-30 payment terms")
facts = await client.logic.query(question="What are Acme Corp's payment terms?")
await client.logic.forget(handles=[fact.handle])
await client.logic.constrain(rule="never schedule calls outside business hours")
snapshot = await client.logic.reflect()

Warden：安全性と検証

check = await client.warden.canary({"action": "delete_all", "scope": "production"})
intent = await client.warden.verify_intent({"action": "transfer_funds", "amount": 50000})
verdict = await client.warden.adjudicate({"claim": "user authorized bulk delete"})
consensus = await client.warden.ensemble({"question": "Is this action safe?", "models": 3})

Flow：ワークフローのオーケストレーション

outcome = await client.flow.run(plan=[
    {"tool": "salesforce@1/get_lead@1", "args": {"id": "123"}},
    {"tool": "quickbooks@1/create_invoice@1", "args": {"$prev.result": True}}
])

matched = await client.flow.route(
    branches=[{"when": "amount > 1000", "then": "approval"}, {"when": "True", "then": "auto"}],
    payload={"amount": 1500}
)

await client.flow.request_approval(action="delete customer record", reason="GDPR request")
await client.flow.request_feedback(missing_fields=["email"], reason="Required for notification")
history = await client.flow.history(limit=20)
details = await client.flow.details(execution_id="exec-abc123")

高階ワークフロー

DataGrout は高階ワークフローに対応しています。これは、フローを他のツール呼び出しのパラメーターとして渡す関数型プログラミングの概念です。これにより、組み合わせ可能で再利用できる自動化を実現できます。

名前付きフロー（スキル）

save_as_skill=True を指定してフローを保存すると、名前付きスキルになり、他のフローから参照できるようになります。

await client.flow.run(
    plan=[
        {"tool": "salesforce@1/get_lead@1", "args": {"id": "$input.lead_id"}},
        {"tool": "data-grout/prism.refract", "args": {"goal": "extract contact info", "payload": "$prev.result"}}
    ],
    save_as_skill=True,
    input_data={"lead_id": "123"}
)

保存したスキルは、別のフローのプラン内で名前を指定して呼び出し、第一級の呼び出し可能オブジェクトとして扱えます。

await client.flow.run(plan=[
    {"tool": "my-saved-skill", "args": {"lead_id": "456"}},
    {"tool": "slack@1/send_message@1", "args": {"channel": "#sales", "text": "$prev.result"}}
])

名前なしフロー（インラインプラン／ラムダ）

$compute を使用すると、ラムダや匿名関数に相当する名前なしフローをインラインで埋め込めます。

await client.flow.run(plan=[
    {"tool": "data-grout/data.query", "args": {"query": "SELECT * FROM leads WHERE status = 'new'"}},
    {"tool": "data-grout/prism.refract", "args": {
        "goal": "enrich each lead",
        "payload": "$prev.result",
        "transform": {"$compute": [
            {"tool": "clearbit@1/enrich@1", "args": {"email": "$item.email"}},
            {"tool": "data-grout/prism.refract", "args": {"goal": "merge enrichment", "payload": "$prev.result"}}
        ]}
    }}
])

条件付きルーティング

flow.route を使うと、述語に基づいてディスパッチできます。条件に応じてペイロードを異なる分岐にルーティングします。

matched = await client.flow.route(
    branches=[
        {"when": "amount > 10000", "then": "high-value-onboarding"},
        {"when": "amount > 1000", "then": "standard-onboarding"},
        {"when": "True", "then": "self-serve"}
    ],
    payload={"amount": 5000, "customer": "Acme Corp"}
)

ヒューマン・イン・ザ・ループ

フローの任意の時点に、承認ゲートやフィードバック要求を挿入できます。

await client.flow.request_approval(
    action="send bulk email to 10,000 contacts",
    reason="Campaign launch requires manager approval",
    details={"template": "spring-promo", "count": 10000}
)

await client.flow.request_feedback(
    missing_fields=["billing_email", "tax_id"],
    reason="Required for invoice generation",
    suggestions={"billing_email": "[email protected]"}
)

成果物とエフェメラル

await client.deliverables.register({"type": "report", "title": "Q1 Summary", "content": report})
items = await client.deliverables.list()
item = await client.deliverables.get("ref-abc123")

cached = await client.ephemerals.list()
entry = await client.ephemerals.inspect("cache-ref-xyz")

汎用フック

名前空間メソッドでまだ対応していない DataGrout ツールには、汎用の dg() フックを使用してください。

facts = await client.dg("invariant.code_lens", {"source": source_code, "language": "python"})

---

ドキュメント

• Python SDK
• TypeScript SDK
• Rust SDK
• Elixir SDK
• Ruby SDK
• DataGrout ライブラリ
• セキュリティ

無料のウェブツール

これらのツールはすべてブラウザー内で動作します。アカウントは不要で、データも保存されません。MCP および AI ツールコミュニティ向けの公共サービスとして提供されています。

• MCP インスペクター — 認証を完全にサポートし、あらゆる MCP サーバーをデバッグできます
• JSONRPC インスペクター — Bearer、OAuth 2.1、mTLS を使用して、あらゆる JSON-RPC 2.0 エンドポイントをテストできます

Labs

DataGrout Labs は、Conduit が連携するシステムに関する研究論文を公開しています。SDK の機能の背景にある「なぜ」を知りたい方は、こちらをご覧ください。

• Cognitive Trust Certificates — エージェントのワークフローに対する形式検証の証明
• Consequential Analysis — 構造的事実と意図の推論による、意味的なコード検証
• Policy & Semantic Guards — 実行時のポリシー適用と動的な編集
• Semio — 型付きツールコントラクトのための意味インターフェース層
• Credits & Virtual Economy — コスト追跡とクレジット見積もりの仕組み
• AI Link Layer — 機械可読なコンテンツ検出プロトコル

---

ライセンス

MIT