Client API

The AdalAgentClient is the full-featured client for multi-query agent sessions. It keeps a persistent connection to the AdaL runtime, letting you send multiple queries, switch models, and resume sessions.

Basic Usage

import anyio
from adal_agent_sdk import AdalAgentClient, AdalAgentOptions


async def main():
    options = AdalAgentOptions(
        workspace="/path/to/project",
        permission_mode="yolo",
    )

    async with AdalAgentClient(options) as client:
        print(f"Session: {client.session_id}")

        await client.query("What files are in this project?")
        async for event in client.receive_events():
            if event.get("type") == "assistant.delta":
                print(event.get("text", ""), end="", flush=True)
            elif event.get("type") in ("command.completed", "command.failed"):
                print()
                break


anyio.run(main)

Multi-Query Conversations

The client maintains conversation context between queries:

async with AdalAgentClient(options) as client:
    # First query
    await client.query("Remember this phrase: pink runtime bridge.")
    async for event in client.receive_events():
        if event.get("type") in ("command.completed", "command.failed"):
            break

    # Second query — context is preserved
    await client.query("What phrase did I ask you to remember?")
    async for event in client.receive_events():
        if event.get("type") == "assistant.delta":
            print(event.get("text", ""), end="", flush=True)
        elif event.get("type") in ("command.completed", "command.failed"):
            break

Resuming Sessions

Resume a previous session by passing its ID:

# Start a session and save its ID
async with AdalAgentClient(options) as client:
    await client.query("Set up the test framework.")
    async for event in client.receive_events():
        if event.get("type") in ("command.completed", "command.failed"):
            break
    session_id = client.session_id

# Later — resume that session
resume_options = AdalAgentOptions(
    workspace="/path/to/project",
    session_id=session_id,
    permission_mode="yolo",
)

async with AdalAgentClient(resume_options) as client:
    await client.query("Now run the tests you set up.")
    async for event in client.receive_events():
        if event.get("type") == "assistant.delta":
            print(event.get("text", ""), end="", flush=True)
        elif event.get("type") in ("command.completed", "command.failed"):
            break

Switching Models

Change the model mid-session:

async with AdalAgentClient(options) as client:
    # Start with one model
    await client.query("Analyze this codebase.")
    async for event in client.receive_events():
        if event.get("type") in ("command.completed", "command.failed"):
            break

    # Switch to a different model
    await client.set_model("claude-sonnet-4-6")

    # Next query uses the new model
    await client.query("Now refactor the auth module.")
    async for event in client.receive_events():
        if event.get("type") in ("command.completed", "command.failed"):
            break

Cancelling a Query

Cancel an in-progress query:

import anyio


async with AdalAgentClient(options) as client:
    await client.query("Refactor all 200 files in src/.")

    async for event in client.receive_events():
        if event.get("type") == "tool.started":
            tool_name = event.get("name", "")
            if tool_name == "bash":
                # Cancel if the agent tries to run shell commands
                await client.cancel()
                break

Context Files and Images

Pass additional context with your query:

await client.query(
    "Review this code for security issues.",
    context_files=["src/auth.py", "src/middleware.py"],
)

# Or include images
await client.query(
    "What does this error screenshot show?",
    images=["screenshot.png"],
)

Handling Events

A complete event handler:

async for event in client.receive_events():
    match event.get("type"):
        case "assistant.delta":
            print(event.get("text", ""), end="", flush=True)
        case "tool.started":
            print(f"\n🔧 {event.get('name')} starting...")
        case "tool.completed":
            print(f"✓ {event.get('name')} done")
        case "reasoning.delta":
            pass  # Extended thinking (usually hidden)
        case "command.completed":
            print("\n✅ Query complete")
            break
        case "command.failed":
            print(f"\n❌ Failed: {event.get('error', 'unknown')}")
            break

Receive Timeout

By default, receive_events() waits up to 5 minutes between messages. For long-running tasks:

# No timeout — wait indefinitely
async for event in client.receive_events(timeout=None):
    ...

# Custom timeout (10 minutes)
async for event in client.receive_events(timeout=600):
    ...

Lifecycle

The client manages the subprocess lifecycle automatically:

# Preferred: async context manager (auto-close)
async with AdalAgentClient(options) as client:
    ...

# Manual lifecycle
client = AdalAgentClient(options)
await client._initialize()
try:
    await client.query("...")
    async for event in client.receive_events():
        ...
finally:
    await client.close()

warning

Always close the client (via context manager or explicit close()). Failing to close leaves the AdaL subprocess running.