DataGrout Conduit SDK

DataGrout Conduit

适用于生产环境的 MCP 客户端，支持 mTLS、OAuth 2.1 和语义发现

可直接替换标准 MCP 客户端。只需替换一行导入代码，Agent 即可获得语义发现、成本追踪、mTLS 身份管理和 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 使用你的 Agent 元数据向 /onramp 发送 POST 请求。
2. DataGrout 返回一个短期有效的会话令牌。
3. SDK 将该令牌提交到 /onramp/complete，以获取 OAuth 凭据和为你配置的 MCP 服务 URL。
4. SDK 在本地生成 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" })

---

身份验证

所有五种 SDK 都支持相同的身份验证方式：

Bearer 令牌

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")

身份自动发现按以下顺序进行：

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. 相对于当前工作目录的 .conduit/

如果一台机器上有多个 Agent，请使用 identity_dir 为每个 Agent 指定独立的证书目录。

DataGrout CA

mTLS 身份是由 DataGrout Certificate Authority（ca.datagrout.ai）签发的 X.509 证书。DataGrout 使用自己的 CA，而不依赖第三方提供商，因为 Agent 身份的需求与浏览器身份不同——Agent 需要快速的程序化签发、自动轮换，以及无需人工流程的机器间信任。

调用 bootstrap_identity 时，会发生以下情况：

1. SDK 在本地生成 ECDSA P-256 密钥对——私钥不会离开你的计算机。
2. 公钥会连同一次性访问令牌或 OAuth 机器令牌一起发送到目标服务的 DG 身份引导端点。
3. DataGrout 会签发证书，将你的公钥绑定到 Substrate 身份（例如 CN=conduit-my-agent），并将其与特定 MCP 服务关联。
4. 签发的证书和 CA 链会返回并保存到磁盘。

此后，每个请求都会出示客户端证书。服务会根据 CA 链和已注册的服务范围身份进行验证——无需令牌、无需在环境变量中存放机密，也无需手动轮换凭据。SDK 会在证书过期前自动续期。

CA 私钥存储在由 HSM 保护的 AWS KMS 密钥中（FIPS 140-2 Level 2），因此签名密钥不会暴露在内存或磁盘中。CA 证书可通过 https://ca.datagrout.ai/ca.pem 公开获取，供需要独立验证证书链的客户端使用。

---

传输模式

DataGrout 通过三种传输方式提供相同的接口——相同的工具、相同的架构和相同的功能。无论选择哪种传输方式，mTLS、OAuth 2.1 和 Bearer 令牌身份验证的工作方式都相同。

传输方式	协议	适用场景
mcp（默认）	基于 Streamable HTTP / SSE 的 MCP	完整的 MCP 协议——支持流式传输、通知，并可直接兼容
jsonrpc	基于 HTTP POST 的 JSON-RPC 2.0	协议更简单、无状态，一次请求对应一次响应
websocket	基于 WebSocket 的 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:// 连接可复用来处理所有并发请求；在同一套接字上并行调用工具时，订阅会持续有效。三种传输方式均支持完整的工具功能——区别在于交付模式：HTTP（拉取）与 WebSocket（推送）。

---

主要功能

智能接口

连接 DataGrout 服务时，默认启用智能接口——将整个工具集替换为两个工具：data-grout@1/discovery.discover@1 和 data-grout@1/discovery.perform@1。你的 Agent 用自然语言描述需求，DataGrout 会在后台处理工具解析、多路复用、数据转换、图表生成等任务。传入 use_intelligent_interface=False 即可选择退出，并查看所有原始工具。

这是对上下文窗口的一项重大优化。你的 Agent 无需推理数百或数千个工具架构，只需查看两个工具。启用后，系统会自动用简化接口替换标准的 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")

认知信任证书

以加密方式证明工作流无循环、类型安全、符合策略且遵守预算。CTC 由颁发 Substrate 身份的同一 DataGrout CA 签名，从 Agent 身份到工作流验证，构建统一的信任链。

---

DataGrout 第一方工具

除了标准的 MCP tools/call 外，每个 SDK 都以带类型的命名空间方法公开完整的 DataGrout 服务 API。特定领域的工具分为六个命名空间：

命名空间	用途
prism	数据转换、制图、渲染、导出、类型桥接
logic	由 Prolog 逻辑单元支持的持久化 Agent 记忆
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（Agent 记忆）

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"}}
])

未命名工作流（内联计划 / Lambda）

你可以使用 $compute 内联嵌入未命名工作流——这相当于 Lambda 或匿名函数：

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]"}
)

Deliverables 与 Ephemerals

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 Library
• 安全

免费 Web 工具

这些工具完全在浏览器中运行——无需账户，也不会存储数据。作为面向 MCP 和 AI 工具社区的公共服务提供。

• MCP 检查器 — 调试任意 MCP 服务，并全面支持身份验证
• JSONRPC 检查器 — 使用 Bearer、OAuth 2.1 和 mTLS 测试任意 JSON-RPC 2.0 端点

实验室

DataGrout Labs 发布有关 Conduit 所交互系统的研究论文。若想了解 SDK 功能背后的“原因”：

• 认知信任证书 — Agent 工作流的形式化验证证明
• 后果分析 — 通过结构事实 + 意图推断进行语义代码验证
• 策略与语义防护 — 运行时策略执行与动态脱敏
• Semio — 面向带类型工具契约的语义接口层
• 额度与虚拟经济 — 成本跟踪与额度估算的工作原理
• AI 链路层 — 机器可读的内容发现协议

---

许可证

MIT