DataGrout Conduit — Python SDK

Production-ready MCP client with mTLS identity, OAuth 2.1, semantic discovery, and cost tracking.

Installation

pip install datagrout-conduit==0.8.0

Quick Start

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

Authentication

Bearer Token

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

OAuth 2.1 (client_credentials)

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

The SDK automatically fetches, caches, and refreshes JWTs before they expire.

OAuth 2.1 (authorization code + PKCE)

client_credentials authenticates a machine, with a secret issued out of band. To authenticate a person — and to reach https://gateway.datagrout.ai/connect, where the server binding is chosen at consent time and lives in the token rather than the URL — run the browser-consent flow once and persist the grant:

from datagrout.conduit import AuthCodeFlow, LoopbackListener

listener = await LoopbackListener.bind()          # 127.0.0.1, OS-chosen port
flow = await AuthCodeFlow.discover("https://gateway.datagrout.ai/connect")
registered = await flow.register("My App", listener.redirect_uri)

url, pending = flow.authorize_url()
print(f"Open this to sign in:\n{url}")            # the SDK never opens a browser

redirect = await listener.wait(timeout=300)
grant = await flow.exchange(pending, redirect.code, redirect.state)

# Persist BOTH: a client id without its redirect URI cannot be reused, because
# the authorization server matches redirect URIs exactly.
save_somewhere({"registered": registered.to_dict(), "grant": grant.to_dict()})

On later runs, skip straight to the grant:

client = Client(
    "https://gateway.datagrout.ai/connect",
    auth={"authorization_code": grant},
)

DataGrout rotates refresh tokens, so a grant that is refreshed and not written back leaves a consumed token on disk. Own the provider when you care:

from datagrout.conduit import AuthCodeProvider

provider = AuthCodeProvider(grant)
client = Client(url, auth={"authorization_code": provider})

# ...periodically, or once on shutdown:
rotated = provider.take_if_dirty()
if rotated is not None:
    save_somewhere({"registered": registered.to_dict(), "grant": rotated.to_dict()})

Where the grant lives is your decision — a keychain, a config file, a vault. The SDK owns its shape and its refresh, and deliberately picks no location. The shape is identical across every conduit SDK, so a grant written by the TypeScript client is readable by this one.

See examples/browser_signin.py for a complete run that registers, signs in, saves, and reuses.

mTLS (Mutual TLS)

After bootstrapping, the client certificate handles authentication at the TLS layer — no tokens needed.

from datagrout.conduit import Client, ConduitIdentity

# Auto-discover from env vars, CONDUIT_IDENTITY_DIR, or ~/.conduit/
client = Client("https://gateway.datagrout.ai/servers/{uuid}/mcp", identity_auto=True)

# Explicit identity from files
identity = ConduitIdentity.from_paths("certs/client.pem", "certs/client_key.pem")
client = Client("...", identity=identity)

# Multiple agents on one machine
client = Client("...", identity_dir="/opt/agents/agent-a/.conduit", identity_auto=True)

Identity Auto-Discovery Order

1. identity_dir option (if provided)
2. CONDUIT_MTLS_CERT + CONDUIT_MTLS_KEY environment variables (inline PEM)
3. CONDUIT_IDENTITY_DIR environment variable (directory path)
4. ~/.conduit/identity.pem + ~/.conduit/identity_key.pem
5. .conduit/ relative to the current working directory

For DataGrout URLs (*.datagrout.ai), auto-discovery runs silently even without identity_auto=True.

Bootstrapping an mTLS Identity

First-run provisioning — generates a keypair, registers with the DataGrout CA, and saves certs locally. After this, the token is never needed again.

# First run: token needed for registration
client = await Client.bootstrap_identity(
    url="https://gateway.datagrout.ai/servers/{uuid}/mcp",
    auth_token="your-access-token",
    name="my-laptop",
)

# Or bootstrap with OAuth 2.1 client_credentials
client = await Client.bootstrap_identity_oauth(
    url="https://gateway.datagrout.ai/servers/{uuid}/mcp",
    client_id="your-client-id",
    client_secret="your-client-secret",
    name="my-laptop",
)

# Subsequent runs: no token needed, mTLS auto-discovered
client = Client("https://gateway.datagrout.ai/servers/{uuid}/mcp")

Semantic Discovery

When use_intelligent_interface is enabled, list_tools() returns only DataGrout's meta-tools. Agents use semantic search instead of enumerating raw integrations:

client = Client("...", use_intelligent_interface=True)

# Semantic search across all connected integrations
results = await client.discover(query="find unpaid invoices", limit=5)

# Direct execution with cost tracking
result = await client.perform(
    tool="salesforce@1/get_lead@1",
    args={"id": "123"},
)

Cost Tracking

Every tool call returns a receipt with credit usage:

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}")
    print(f"Savings: {meta.receipt.savings}")

Transports

# MCP (default) — full MCP protocol over Streamable HTTP
client = Client(url)

# JSONRPC — lightweight, stateless, same tools and auth
client = Client(url, transport="jsonrpc")

# WebSocket — bidirectional push; requires pip install 'datagrout-conduit[ws]'
client = Client("wss://gateway.datagrout.ai/servers/{uuid}/ws", transport="websocket")

WebSocket transport

The WebSocket transport uses the datagrout-jsonrpc.v1 subprotocol over a single persistent wss:// connection. All concurrent requests are multiplexed on that connection; responses are correlated by JSON-RPC id via asyncio.Future with no head-of-line blocking.

from datagrout.conduit import Client

async with Client(
    "wss://gateway.datagrout.ai/servers/{uuid}/ws",
    auth={"bearer": "your-token"},
    transport="websocket",
) as client:
    # Subscribe to server-pushed events
    sub = await client.subscribe("agents.my-agent-id.events")

    async for event in sub:
        print(f"{event.event}: {event.data}")

    await client.unsubscribe(sub.id)

Supported topics:

You can also call await sub.recv() for manual, one-event-at-a-time consumption:

event = await sub.recv()
print(event.event, event.data)

Reconnection: after a disconnect, send_request raises RuntimeError("WS transport not connected"). Re-call connect() and re-subscribe — subscriptions do not survive reconnects in v0.4.

API Reference

Client Options

Client(
    url: str,
    auth: dict = None,                    # {"bearer": ...}, {"client_credentials": {...}},
                                          # or {"authorization_code": Grant | AuthCodeProvider}
    transport: str = "jsonrpc",           # "jsonrpc", "mcp", or "websocket"
    use_intelligent_interface: bool = False,
    identity: ConduitIdentity = None,     # explicit mTLS identity
    identity_auto: bool = False,          # auto-discover identity
    identity_dir: str = None,             # custom identity directory
    disable_mtls: bool = False,           # opt out of mTLS auto-discovery
    client_id: str = None,               # OAuth shorthand
    client_secret: str = None,           # OAuth shorthand
)

Standard MCP Methods

DataGrout Extensions

Bootstrap Methods

Requirements

• Python 3.10+
• httpx (for JSONRPC transport)
• mcp package (optional, for MCP transport mode)

License

MIT