Conduit SDK

Drop-in MCP client for Python, TypeScript, and Rust

Conduit replaces your existing MCP client with a single import swap. Your agent code works unchanged. The SDK transparently routes every call through the DataGrout intelligence layer, giving you discovery, planning, policy enforcement, and cost tracking without rewriting anything.

Installation

Python

pip install datagrout-conduit

TypeScript / Node.js

npm install datagrout-conduit

Rust

cargo add datagrout-conduit

Connect to Your Server

You need your server URL (from the dashboard) and an authentication credential.

Python

from datagrout.conduit import Client

client = Client(
    "https://gateway.datagrout.ai/servers/YOUR_SERVER_ID/mcp",
    auth={"bearer": "dg_your_token_here"},
)
await client.connect()

TypeScript

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

const client = new Client({
    url: 'https://gateway.datagrout.ai/servers/YOUR_SERVER_ID/mcp',
    auth: { bearer: 'dg_your_token_here' },
});
await client.connect();

Rust

use datagrout_conduit::{ClientBuilder, Transport};

let client = ClientBuilder::new()
    .url("https://gateway.datagrout.ai/servers/YOUR_SERVER_ID/mcp")
    .auth_bearer("dg_your_token_here")
    .build()?;

client.connect().await?;

Standard MCP Operations

Conduit supports the full MCP protocol surface. These calls behave identically to any MCP client.

List Tools

tools = client.list_tools()

for tool in tools:
    print(tool.name, tool.description)

Call a Tool

result = client.call_tool(
    "salesforce@1/get_leads@1",
    {"limit": 10}
)

DataGrout Extensions

Beyond standard MCP, Conduit exposes three methods that access the intelligence layer directly.

discover()

Find tools by meaning, not by name. Returns ranked matches based on semantic similarity to your query.

results = client.discover(
    query="create an invoice",
    limit=5,
    min_score=0.5
)

for match in results:
    print(match.tool.name, match.score)

Use goal instead of query when you want multi-step planning:

plan = client.discover(
    goal="sync qualified leads to invoices"
)

Filter to a specific integration:

results = client.discover(
    query="list customers",
    integration="quickbooks"
)

guide()

Start an interactive decision tree to build a workflow step-by-step.

session = client.guide(goal="export invoices to PDF")

# System returns options at each step
print(session.message)
for option in session.options:
    print(option.id, option.label)

# Select a choice by ID
session = client.guide(choice="1.2")

See Guide Mode for the full walkthrough.

perform()

Execute a tool directly through the intelligence layer. Equivalent to call_tool but routes through policy enforcement, cost tracking, and optional output transformations.

result = client.perform(
    "quickbooks@1/create_invoice@1",
    {"customer_id": "123", "amount": 1500.00}
)

Cost Tracking

Every tool call returns a receipt with credit costs. Use extract_meta() to access it.

Python

from datagrout.conduit import extract_meta

result = await client.call_tool("salesforce@1/get_leads@1", {"limit": 10})
meta = extract_meta(result)

if meta:
    print(meta.receipt.net_credits)
    print(meta.receipt.savings)

TypeScript

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

const result = await client.callTool('salesforce@1/get_leads@1', { limit: 10 });
const meta = extractMeta(result);

if (meta) {
    console.log(meta.receipt.netCredits);
    console.log(meta.receipt.savings);
}

Rust

let result = client.call_tool("salesforce@1/get_leads@1", json!({"limit": 10})).await?;
let meta = client.extract_meta(&result)?;
println!("Cost: {} credits", meta.receipt.actual_credits);

See Credits and Receipts for receipt fields and BYOK discounts.

Transport Options

Conduit supports two transport protocols. Both provide identical access to the intelligence layer.

Transport Protocol Streaming Use When
Transport.Mcp JSON-RPC 2.0 over SSE Yes Your agent framework supports MCP
Transport.JsonRpc JSON-RPC 2.0 over HTTP POST No Legacy systems, simple HTTP clients 
# MCP transport (streaming, persistent connection)
client = ClientBuilder().transport(Transport.Mcp).build()

# JSON-RPC transport (stateless HTTP POST)
client = ClientBuilder().transport(Transport.JsonRpc).build()

Authentication

Conduit supports three authentication methods: bearer tokens, OAuth 2.1, and mutual TLS.

Bearer Token

The simplest option. Create a token in your server settings.

Python

client = Client(
    "https://gateway.datagrout.ai/servers/YOUR_UUID/mcp",
    auth={"bearer": "your-token"},
)

Rust

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

OAuth 2.1 (client_credentials)

For machine-to-machine authentication. The SDK handles token exchange and refresh automatically.

Python

client = Client(
    "https://gateway.datagrout.ai/servers/YOUR_UUID/mcp",
    client_id="your-client-id",
    client_secret="your-client-secret",
)

Rust

let client = ClientBuilder::new()
    .url("https://gateway.datagrout.ai/servers/YOUR_UUID/mcp")
    .auth_client_credentials("your-client-id", "your-client-secret")
    .build()?;

Mutual TLS (mTLS)

Certificate-based authentication for production agents. After bootstrapping, no tokens or secrets are needed — the client certificate is the credential.

Bootstrap once, connect forever:

# First run: register identity with the DataGrout CA
client = await Client.bootstrap_identity(
    url="https://gateway.datagrout.ai/servers/YOUR_UUID/mcp",
    auth_token="your-access-token",
    name="my-agent",
)

# All subsequent runs: mTLS auto-discovered from ~/.conduit/
client = Client("https://gateway.datagrout.ai/servers/YOUR_UUID/mcp")

Auto-discovery searches:

  1. CONDUIT_MTLS_CERT / CONDUIT_MTLS_KEY env vars (inline PEM)
  2. CONDUIT_IDENTITY_DIR env var (directory path)
  3. ~/.conduit/
  4. .conduit/ relative to cwd

Multiple agents on one machine:

client = Client(
    "https://gateway.datagrout.ai/servers/YOUR_UUID/mcp",
    identity_dir="/opt/agents/agent-a/.conduit",
    identity_auto=True,
)

See Authentication for the full reference.

Full Example

Connect, discover a tool, execute it, and check the cost.

from datagrout.conduit import Client, extract_meta

async with Client(
    "https://gateway.datagrout.ai/servers/YOUR_SERVER_ID/mcp",
    auth={"bearer": "dg_your_token_here"},
) as client:
    # Find the right tool by intent
    matches = await client.discover(query="list recent invoices", limit=3)
    tool = matches[0].tool

    # Execute it
    result = await client.call_tool(tool.name, {"limit": 25})

    # Check cost
    meta = extract_meta(result)
    if meta:
        print(f"Credits: {meta.receipt.net_credits}, savings: {meta.receipt.savings}")

Related