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.1

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

identity_dir option (if provided)
CONDUIT_MTLS_CERT + CONDUIT_MTLS_KEY environment variables (inline PEM)
CONDUIT_IDENTITY_DIR environment variable (directory path)
~/.conduit/identity.pem + ~/.conduit/identity_key.pem
.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:

Topic	Fires when
`agents.<agent_id>.events`	Agent lifecycle events (plan started, IC completed, grounding failed, …)
`tools.<tool_name>.results`	A specific tool call completes
`tasks.<task_id>.*`	Long-running background task transitions
`flows.<flow_id>.*`	`flow.into` progress and completion
`governor.<server_uuid>`	Governor percept events (file change, schedule, webhook)

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

Method	Description
`list_tools()`	List available tools
`call_tool(name, args)`	Execute a tool
`list_resources()`	List resources
`read_resource(uri)`	Read a resource
`list_prompts()`	List prompts
`get_prompt(name, args)`	Get a prompt

DataGrout Extensions

Method	Description
`discover(query, limit, integrations)`	Semantic tool search
`perform(tool, args, demux)`	Direct tool execution with tracking
`perform_batch(calls)`	Parallel tool execution
`guide(goal, policy, session_id)`	Guided multi-step workflow
`flow_into(plan, ...)`	Workflow orchestration
`prism_focus(data, lens)`	Data transformation via Prism lens
`estimate_cost(tool, args)`	Pre-execution credit estimate

Bootstrap Methods

Method	Description
`Client.bootstrap_identity(url, auth_token, name)`	Bootstrap mTLS with access token
`Client.bootstrap_identity_oauth(url, client_id, client_secret, name)`	Bootstrap mTLS with OAuth 2.1

Requirements

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

License

MIT

Navigation Menu

Repository navigation

DataGrout Conduit — Python SDK

Installation

Quick Start

Authentication

Bearer Token

OAuth 2.1 (client_credentials)

OAuth 2.1 (authorization code + PKCE)

mTLS (Mutual TLS)

Identity Auto-Discovery Order

Bootstrapping an mTLS Identity

Semantic Discovery

Cost Tracking

Transports

WebSocket transport

API Reference

Client Options

Standard MCP Methods

DataGrout Extensions

Bootstrap Methods

Requirements

License

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

DataGrout Conduit — Python SDK

Installation

Quick Start

Authentication

Bearer Token

OAuth 2.1 (client_credentials)

OAuth 2.1 (authorization code + PKCE)

mTLS (Mutual TLS)

Identity Auto-Discovery Order

Bootstrapping an mTLS Identity

Semantic Discovery

Cost Tracking

Transports

WebSocket transport

API Reference

Client Options

Standard MCP Methods

DataGrout Extensions

Bootstrap Methods

Requirements

License