API Reference

Module: adal_agent_sdk

All public exports are available from the top-level package:

from adal_agent_sdk import (
    query,
    AdalAgentClient,
    AdalAgentOptions,
    PermissionResult,
    ToolPermissionContext,
    SubprocessTransport,
    SdkError,
    AdalConnectionError,
    QueryError,
    ProtocolError,
    SdkRuntimeError,
)

---

query()

async def query(
    prompt: str,
    options: AdalAgentOptions | None = None,
    **kwargs,
) -> AsyncIterator[dict[str, Any]]

One-shot convenience function. Spawns a client, sends the prompt, yields events, and cleans up.

Parameters:

Parameter	Type	Description
prompt	str	The user prompt to send
options	AdalAgentOptions | None	Pre-built options object
**kwargs		Shorthand for AdalAgentOptions fields (e.g., workspace=".")

Yields: dict[str, Any] — SDK event dictionaries.

Raises: ValueError if both options and keyword arguments are provided.

Example:

async for event in query("Fix the bug", workspace=".", permission_mode="yolo"):
    print(event)

---

AdalAgentClient

class AdalAgentClient:
    def __init__(self, options: AdalAgentOptions): ...

Persistent client for multi-query sessions. Use as an async context manager.

Properties

Property	Type	Description
session_id	str | None	Active session ID (available after init)

Methods

query(input, **kwargs)

async def query(self, input: str, **kwargs) -> None

Send a query to the agent. Events are received via receive_events().

Parameter	Type	Description
input	str	User input text
images	list[str]	Optional image paths or data URIs
display_text	str	Optional display text override
context_files	list[str]	Optional additional context file paths

receive_events(timeout)

async def receive_events(
    self, timeout: float | None = 300.0
) -> AsyncIterator[dict[str, Any]]

Yield events from the active query until completion. Handles control frames (approval callbacks) transparently.

Parameter	Type	Default	Description
timeout	float | None	300.0	Max seconds between messages. None = no timeout

set_model(model)

async def set_model(self, model: str) -> dict[str, Any]

Switch the active model mid-session.

set_permission_mode(mode)

async def set_permission_mode(self, mode: str) -> dict[str, Any]

Change the permission mode ("default", "acceptEdits", or "yolo").

cancel()

async def cancel(self) -> None

Cancel the active query.

close()

async def close(self) -> None

Shut down the runtime and close the connection.

---

AdalAgentOptions

@dataclass
class AdalAgentOptions:
    workspace: str | Path | None = None
    model: str | None = None
    session_id: str | None = None
    permission_mode: str | None = None
    auth_token: str | None = None

    enabled_default_tools: list[str] | None = None
    disabled_default_tools: list[str] | None = None
    thinking_effort: str | None = None
    prompt_file: str | Path | None = None
    can_use_tool: CanUseTool | None = None
    runtime_path: str | Path | None = None

Field	Type	Description
workspace	str | Path | None	Workspace root (defaults to cwd)
model	str | None	Model ID for startup
session_id	str | None	Session ID to resume (None = new session)
permission_mode	str | None	"default", "acceptEdits", or "yolo"
auth_token	str | None	JWT for platform auth (omit for cached)

| enabled_default_tools | list[str] \| None | Positive set — ONLY these tools are available (cannot combine with disabled_default_tools) | | disabled_default_tools | list[str] \| None | Core tool groups/names to disable entirely (invisible + unexecutable) | | thinking_effort | str \| None | "low", "medium", "high", or "max" | | prompt_file | str \| Path \| None | Path to a file whose contents replace the default system prompt | | can_use_tool | CanUseTool \| None | Async permission callback | | runtime_path | str \| Path \| None | Path to adal binary (defaults to PATH) |

---

PermissionResult

Factory class for building permission callback return values.

PermissionResult.allow(updated_input=None)

@staticmethod
def allow(updated_input: dict | None = None) -> PermissionResultAllow

Allow the tool call. Optionally provide modified input.

PermissionResult.deny(message="")

@staticmethod
def deny(message: str = "") -> PermissionResultDeny

Deny the tool call with an optional reason.

---

ToolPermissionContext

@dataclass
class ToolPermissionContext:
    tool_call_id: str
    confirmation: dict[str, Any] | None = None
    display: dict[str, Any] | None = None

Field	Type	Description
tool_call_id	str	Unique identifier for this tool call
confirmation	dict | None	Confirmation metadata (diff, prompt)
display	dict | None	Display metadata

---

SubprocessTransport

class SubprocessTransport:
    def __init__(self, runtime_path=None, cwd=None): ...

Low-level transport layer. Manages the AdaL subprocess and NDJSON stdio communication. Typically not used directly — AdalAgentClient handles this internally.

---

Exceptions

Exception	Description
SdkError	Base exception for all SDK errors
AdalConnectionError	Failed to connect to the AdaL runtime
QueryError	Error during query execution
ProtocolError	Wire protocol violation
SdkRuntimeError	Runtime-level error from the AdaL process

All exceptions inherit from SdkError:

from adal_agent_sdk import SdkError, QueryError

try:
    async for event in query("..."):
        pass
except QueryError as e:
    print(f"Query failed: {e}")
except SdkError as e:
    print(f"SDK error: {e}")

---

Type Aliases

# Permission callback signature
CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResultAllow | PermissionResultDeny],
]

---

Environment Variables

Variable	Description
ADAL_RUNTIME_PATH	Override the path to the adal binary

---

Version

from adal_agent_sdk import __version__
print(__version__)  # "0.1.0"