Python

Python 3.10+ clients for an authenticated AgenticDriver host. Provider API keys and subscription credentials stay on that host. Choose a provider instance and model explicitly; clients never switch them or retry a request automatically.

Download the reviewed wheel from the current release. PyPI organization approval is pending. Install the public archive in your application's own environment:

python -m pip install https://github.com/agenticdriver/agenticdriver/releases/download/v0.2.0-rc.2/agenticdriver-0.2.0rc2-py3-none-any.whl
# Native asyncio support:
python -m pip install 'agenticdriver[async] @ https://github.com/agenticdriver/agenticdriver/releases/download/v0.2.0-rc.2/agenticdriver-0.2.0rc2-py3-none-any.whl'

The base installation uses the standard-library HTTP transport and typing_extensions. The async extra adds HTTPX. Importing the package, using types, or running the synchronous client does not require HTTPX.

Synchronous execution#

Optional detached jobs expose submit_job, read_job, cancel_job and job_events on both clients. Their JobSubmit, JobIdentity, JobInfo, JobEventsRequest and JobEventPage types are public. Aborting a job HTTP request does not cancel accepted background work; use cancel_job explicitly. Reconnect with the last accepted nextCursor. Pages never dispatch tools.

from agenticdriver import AgenticClient, RunRequest

request: RunRequest = {
    "provider": "my-instance", "model": "explicit-model", "input": "Summarize this",
}
with AgenticClient("http://127.0.0.1:7433", token="YOUR_DRIVER_TOKEN") as client:
    print(client.providers(refresh=True))  # Read-only discovery, no generation.
    result = client.run(**request)
    print(result["text"])
    with client.stream(**request) as events:
        for event in events:
            if event["type"] == "text.delta":
                print(event["text"], end="", flush=True)

RunStream is an iterator and context manager. Completion, a protocol error, close(), or leaving with closes the response. Breaking iteration alone does not imply ownership cleanup: use with or call close() explicitly. Closing a client closes its outstanding streams. Do not share a synchronous stream across threads; use asyncio task cancellation to interrupt an active receive.

Native asyncio execution#

import asyncio
from agenticdriver import AsyncAgenticClient, RunRequest

async def main() -> None:
    request: RunRequest = {
        "provider": "my-instance", "model": "explicit-model", "input": "Summarize this",
    }
    async with AsyncAgenticClient("https://driver.example", token="YOUR_DRIVER_TOKEN") as client:
        result = await client.run(**request)
        print(result["text"])
        async with client.stream(**request) as events:
            async for event in events:
                if event["type"] == "run.completed":
                    print(event["result"]["usage"])

asyncio.run(main())

There is no worker-thread wrapper. One asyncio task reads each stream; independent streams can share a client on the same event loop. Cancelling the consuming task propagates asyncio.CancelledError and closes its response. await stream.aclose() or await client.aclose() also interrupts outstanding stream reads. Read streams inside async with, including when stopping early or raising an application error. Do not depend on garbage collection to cancel work.

Both styles expose protocol, providers, run, stream, ingest_context, index_context, search_context, and delete_context; await the async methods except stream, which returns an async iterator/context manager.

Types, errors, and connection policy#

RunRequest, RunResult, Usage, ProviderInfo, ProtocolInfo, and the tagged RunEvent union are exported from agenticdriver and agenticdriver.models. Context, retrieval, and ingestion models are also exported. They are TypedDict wire objects, so existing dictionary code works. run(**request) and stream(**request) check required and optional keyword fields with mypy; event types narrow on event["type"]. Runtime response validation remains enabled. DriverError exposes code, retryable, and optional outcome == "uncertain". An uncertain outcome needs reconciliation before repeating external effects. Streaming failed/cancelled terminal events retain their typed error payload.

Remote endpoints require verified HTTPS. HTTP is permitted only on loopback. Pass ca_file="/path/to/private-ca.pem" for a private CA; hostname verification stays enabled. Redirects are not followed. The async client does not inherit environment proxies, .netrc, or ambient credentials. An injected HTTPX transport is caller-owned policy for tests or custom networking; it must verify TLS appropriately when it performs real requests.

Execution has no default total deadline or inactivity timeout. A request's idleTimeoutMs opts into host-side inactivity detection based on real model/tool/ context progress. Heartbeat comments do not reset it. Constructor io_timeout defaults to None for run/retrieval/ingestion calls. An explicit positive value sets socket/HTTP I/O timeouts, which cannot distinguish reasoning from a stalled network and may interrupt a healthy quiet run. discovery_timeout defaults to 10 seconds for read-only discovery and can be changed or disabled. These are separate from the host execution policy. An application can use its own asyncio cancellation or asyncio.wait_for when it needs an explicit deadline.

The async transport disables HTTPX's default network timeout for execution and uses its response close lifecycle.

Verification#

npm run test:clients builds and installs the actual language packages, compiles examples and runs pure contracts. It does not simulate connections or model answers. Use real connection checks for provider execution.

Interactive approvals#

Select approvals={"mode": "interactive", "idlePolicy": "pause"} in stream(). On an approval.requested event, review event["approval"]["call"] and submit decide_approval({"approvalId": approval["approvalId"], "runId": approval["runId"], "call": approval["call"], "decision": "approve"}); await this on AsyncAgenticClient. Decisions may also be deny or cancel. Typed policies, requests, decisions and resolutions are exported from agenticdriver.

The host must enable this feature and grant approveTools. run rejects interactive mode because it cannot deliver review requests. There is no default approval expiry; choose expiresAfterMs / the typed equivalent when needed. Decisions are single-use and are never retried automatically. A receipt does not confirm a tool effect. See the approval contract.

Functions in your application#

Pass typed ApplicationToolDefinition dictionaries in applicationTools, and select the same names in tools. Use stream() and dispatch only a tool.execution.requested event to your local function. From event["execution"], copy executionId, runId and call["id"] (as callId) into an identity. report_tool_progress(identity) reports real work; complete_tool({**identity, "output": output}) submits the JSON result once. Await both methods on AsyncAgenticClient. For callback failures, use "error": "APPLICATION_TOOL_FAILED" instead of output; do not transmit exceptions.

The host must enable application tools and grant the named functions. Review is required by default. Closing the stream invalidates pending tickets; cancellation of work inside your own function is cooperative. There is no default inactivity deadline or automatic retry. See the full contract.

Conversation sessions#

Create/read/delete methods and typed session models support explicit conversation continuation. Pass only the returned session ID and revision on each run, then use the next revision from the successful result. Provider/account/model changes require an explicitly created conversation with exported visible history. Host opt-in, account binding and per-operation token grants are required. Idle retention pauses during active work; no default execution deadline is introduced. See the session contract and binding examples.

Provider settings component#

The provider and connection panel supplies provider/model settings, remote management and local/remote setup through the existing SDK client. TypeScript, Python, Go and Rust packages share the same embeddable browser component and provide native backend bindings. These additions are available on sdk-roadmap and are not in the published 0.1.0 packages.