Python SDK reference: ask, next, AgentResponse

Import common interfaces from dualeai; specialized module paths are shown below. In signatures, * starts the keyword-only arguments. These APIs are in Public Preview.

ask

Submit a Task and return an AgentResponse once the SDK’s circuit breaker allows the submission. Duale AI has not necessarily accepted the request when the call returns.

async def ask(
    action: str,
    capabilities: list[Capability] | None = None,
    res: type | None = None,
    response_format: ResponseFormat | None = None,
    routing: RoutingPolicy | None = None,
    *,
    priority: PriorityClass = PriorityClass.standard,
    streaming: bool = False,
    deadline: datetime | None = None,
    attachments: list[PreparedAttachment] | None = None,
    task_id: str | None = None,
    sdk: DualeAISDK,
) -> AgentResponse

next

Start a continuation Task that inherits the Task history of an earlier Task, while that Task runs or after its outcome.

async def next(
    self,
    message: str,
    *,
    priority: PriorityClass = PriorityClass.standard,
    res: type[T] | None = None,
    deadline: datetime | None = None,
    response_format: ResponseFormat | None = None,
    streaming: bool | None = None,
    routing: RoutingPolicy | None = None,
    attachments: list[PreparedAttachment] | None = None,
    task_id: str | None = None,
) -> AgentResponse[T]

None preserves the parent’s response format and streaming choice. The deadline defaults to 30 minutes from submission, and priority defaults independently to standard. res= is not inherited: when response_format is omitted, res= sends its JSON Schema as the continuation’s format. With neither, the continuation keeps the parent’s format and performs no local validation.

next() sends the continuation as soon as Duale AI has accepted the parent’s request. It does not wait for the parent’s outcome or validate the parent’s result. Call await response.model() first if you need that result. The returned AgentResponse.task_id is the task_id argument, or a new identifier generated before submission. The parent’s identifier stays unchanged. See Continuation Tasks.

The annotations still use the parent’s AgentResponse[T], so they do not reflect these runtime type changes. The method never raises the parent’s outcome. A permission refusal is an immediate HTTP answer; awaiting the returned response’s model() raises DualeAIAuthError with AUTHORIZATION_FAILED.

If the parent is not ready yet, the continuation waits. When a continuation waits or fails states the rules, and Task outcomes lists PARENT_TASK_NOT_FOUND and PARENT_TASK_UNAVAILABLE, which model() raises as DualeAIError. Cancelling a caller waiting in next() before it sends leaves the parent running.

AgentResponse

Returned by ask() and response.next().

response.task is an asyncio.Task that submits the request and observes events locally. Cancelling it, or a caller waiting in model(), ends that observation without sending a Stop. To pause waiting, keep a task for model(), protect it with asyncio.shield(), and await it later. Cancellation can also propagate through cache_hit() and llm_metrics(). Check success with model(); the metadata methods are not outcome checks.

stop_task

Request a Stop for a Task. Duale AI processes it asynchronously.

async def stop_task(self, task_id: str, reason: str) -> TaskStopAccepted

TaskStopAccepted confirms that the caller has permission to stop the Task; it does not prove that the Task exists or that it stopped. Keep observing the original response: model() raises TaskStoppedError with your reason if the Stop takes effect. Otherwise, the Task keeps the outcome it reached first. stream() yields no separate terminal Stop event.

from dualeai import TaskStoppedError

response = await ask("Analyse this contract", sdk=sdk)
await response.stop(reason="Wrong document supplied")

try:
    await response.model()
except TaskStoppedError as stopped:
    print(stopped.reason)

Use response.stop(reason) for a response you hold, or sdk.stop_task(task_id, reason) with a saved ID, including one submitted by another process. An Agent Identity has permission to stop its own Tasks by default, so these calls work with its API token out of the box. Without agent:stop_task on the Task, the call immediately raises DualeAIAuthError (AUTHORIZATION_FAILED, 403). An allowed request receives acceptance (202), even if the Task ended or does not exist. A Stop sent before the Task starts, including to a continuation still waiting for its parent, is applied only if the Task starts within a limited period. Confirm the Stop through the Task outcome.

Stop the Task you intend to stop explains which work a Stop covers, how it affects continuations, and what it cannot undo.

Task actions

A call answered with a Task’s event stream (ask, next, or Tool Result submission) also needs agent:read_task on that Task. A permission refusal is answered at once with AUTHORIZATION_FAILED. See Tenant access policies for resource patterns and {self}, and Default policies for the supplied grants.

DualeAISDK

The client. Use it as an async context manager, or call serve() to publish tools.

class DualeAISDK:
    def __init__(
        self,
        config: DualeAIConfig | None = None,
        *,
        agent_id: str | None = None,
        max_jobs: int = 100,
        job_timeout: int = 1800,
        auto_start: bool = True,
        backpressure_config: BackpressureConfig | None = None,
        max_concurrent_tools: int | None = None,
        graceful_shutdown_timeout: float = 0.0,
        transport: HTTPTransportProtocol | None = None,
    ) -> None: ...

    async def serve(self, *, stop_event: asyncio.Event | None = None) -> None: ...

config defaults to a DualeAIConfig() built from the environment. max_jobs sets the default concurrency for cached activities and registered tools. job_timeout limits the wait for scheduled activity work; see Activity caching for its scope. max_concurrent_tools overrides only the registered-tool concurrency and otherwise defaults to max_jobs.

On shutdown, graceful_shutdown_timeout allows that many seconds for in-flight Tool Results. The 0.0 default cancels the SDK waits immediately; a synchronous callable already running in a worker thread can continue. auto_start defaults to True on the constructor. The create_sdk() helper and MockSDK default it to False.

For Task submissions, use async with create_sdk() as sdk:. To publish registered Tools for agent_id, call serve(). It maintains heartbeats and blocks until shutdown. Use stop_event= for cooperative shutdown.

Tests can pass transport= to inject an HTTPTransportProtocol. dualeai.testing provides the simpler keyed-response and in-memory Library helpers.

Close the SDK after its work finishes

Keep the async context open until you have consumed every response. On exit, await sdk.cleanup() cancels Task observers, stops Tool handling and heartbeats, and closes network and cache resources. Request a Stop explicitly if needed; cleanup does not send one.

await sdk.start() registers the current Tools and starts heartbeats. Repeated calls do not start another lifecycle. To control startup, construct with auto_start=False, declare all Tools, then call start() or serve(). serve() also runs cleanup when it exits. See Authoring tools for the complete serving example.

create_sdk

create_sdk(**kwargs) builds a DualeAISDK from keyword arguments, falling back to DUALEAI_ environment variables. It defaults to auto_start=False.

from dualeai import create_sdk

sdk = create_sdk(agent_id="agent_security_operations")

Accepted keys: token, endpoint, tenant_id, agent_id, redis_url, sqlite_path, debug, max_jobs, job_timeout, auto_start.

Use DualeAISDK(...) directly for transport, backpressure_config, max_concurrent_tools, or graceful_shutdown_timeout. Pass a DualeAIConfig for nested observability settings. These are not create_sdk() keywords.

DualeAIConfig

DualeAIConfig reads DUALEAI_ environment variables; explicit values override them. SDK configuration lists every variable, its default, and its rules, and also covers logging, telemetry export, and local health.

PriorityClass

Import PriorityClass from dualeai. ask() and response.next() accept the keyword-only priority argument. Its values are PriorityClass.standard and PriorityClass.fast; omitting it selects standard. None and integers are not accepted classes.

Standard favors throughput; fast favors lower waiting time for the same work and uses the Fast Task price. Model selection, Tool behavior, and the Task deadline keep their own contracts. Fast does not guarantee an end-to-end response time.

A Task keeps the class it started with across retries. Duale AI charges a Task once, at its class’s price, when it starts working on it. A request that ends before work starts has no Task charge, and retries or work inside the Task add none. Each continuation is a separate Task with its own class and charge; omitting priority selects standard even after a fast parent.

Excerpt inside an open SDK context:

from dualeai import PriorityClass, ask

response = await ask("Summarize the supplied report", priority=PriorityClass.fast, sdk=sdk)
await response.model()
follow_up = await response.next("List the unresolved questions", priority=PriorityClass.fast)
await follow_up.model()

Omit priority from next() to request a standard continuation. Scheduling class is separate from the model-selection preferences in RoutingPolicy.

RoutingPolicy

All fields are optional, soft preferences. Even required_capabilities allows fallback when no model meets the full request.

Omitting a field leaves it out of the submitted policy. For a new Task, Duale AI resolves omitted cost_sensitivity to 0.1. Set it to None or 0 to remove the general preference for lower model prices, or choose another value from 0 to 1. Duale AI applies its own defaults to the other omitted fields. A retry keeps the policy the Task started with.

Routing contract and limits defines what these values mean, how routing handles omitted preferences, and which controls to use for hard requirements.

An omitted field reads back as None on the RoutingPolicy instance. Explicit None is accepted only for the four nullable numeric fields; omit capability lists instead of setting them to None.

For a continuation, each omitted field inherits from the parent, even when you pass RoutingPolicy() or a partial policy. Set a nullable numeric field to None to clear the inherited preference, or a capability list to [] to clear the inherited list. Duale AI’s own needs for each part of the work still apply, so these overrides do not guarantee the same effective preferences. What a continuation keeps explains the scope of inheritance.

ResponseFormat

Use a predefined string—text, json, markdown, html, xml, yaml, or csv—or an object with a json_schema and an optional description. Pass a predefined format as its string, for example response_format="json". For validation against a strict schema, construct the object form:

from dualeai.models.response_format import JsonSchemaResponseFormat

response_format = JsonSchemaResponseFormat(
    json_schema={"type": "object", "properties": {"total": {"type": "number"}}, "required": ["total"]},
    description="Invoice totals",
)

Capability

problem_solving, math, code, agentic, long_context, finance, cyber_defensive, cyber_offensive, medical, legal.

A capability describes the work a Task needs. Routing uses available benchmark evidence to inform that preference; coverage and evaluation conditions vary by model and benchmark. Task capabilities and benchmark evidence defines each value, gives examples, and names its associated benchmarks.

Use these values in RoutingPolicy. They are best-effort preferences, not safety controls or permission to perform a use prohibited by the terms of use.

Decorators: @tool and @activity

Use these signatures to declare Tools and cached activities:

@tool(*, sdk: DualeAISDK, description: str, timeout: timedelta,
      retries: int = 0, error_transform: Callable[[Exception], str] | None = None)

@activity(cache_ttl: timedelta | None = None, max_retries: int = 3, *, sdk: DualeAISDK)

@tool registers a function that a Task can call; see Authoring tools. @activity caches an async function called by your application; it does not publish a Tool. Both decorators require sdk= when the function is declared. Activity caching explains storage, expiry, and retries.

Tool context fields

Inside a Tool Call, current_tool_context() returns a ToolContext with tool_call_id, attempt, task_id, and deadline_at. It also provides remaining_seconds() and is_expiring(). Outside a Tool Call, the function returns None. See Authoring tools.

Streaming events

response.stream() yields content events in arrival order, which is not always the answer’s order.

Show only the highest generation, with its deltas sorted by sequence; Stream an answer shows the rule. Take the final result from await response.model(), not from the streamed text.

Attachments

sdk.prepare_attachments(files) accepts a list of (Path, description) tuples and returns PreparedAttachment references for non-empty regular files. It reads their sizes, not their contents. Descriptions have at most 500 characters. Keep each file unchanged between preparation and upload.

await sdk.upload_attachments(task_id, attachments, agent_id=None) uploads into the task’s Library and returns dict[str, LibraryDocumentCreateResponse], keyed by PreparedAttachment.key. Omit agent_id to use sdk.agent_id, configured by the constructor or DUALEAI_AGENT_ID. A non-empty upload without either identifier raises ValueError; an empty attachment list returns {} without a request.

Pass the same prepared references to ask(attachments=...). Upload batches can partially succeed; follow Attach documents and its recovery link before retrying.

LibrariesClient

sdk.libraries manages persistent Library metadata and documents. Before any operation, set tenant_id through DualeAIConfig(tenant_id=...) or DUALEAI_TENANT_ID. The server identifies the caller from DUALEAI_TOKEN and decides each operation from that caller’s access to the Library’s path.

class LibrariesClient:
    async def create(self, request: LibraryCreateRequest) -> LibraryWithRevision: ...
    async def list(self) -> LibraryListResponse: ...
    async def get(self, request: LibraryGetRequest) -> LibraryWithRevision: ...
    async def update(self, request: LibraryUpdateRequest) -> LibraryWithRevision: ...
    async def delete(self, request: LibraryDeleteRequest) -> None: ...

    async def upload(
        self,
        library_id: str | UUID,
        attachments: list[PreparedAttachment],
    ) -> dict[str, LibraryDocumentCreateResponse]: ...
    async def list_documents(
        self,
        request: LibraryDocumentListRequest,
    ) -> LibraryDocumentPage: ...
    async def get_document(
        self,
        request: LibraryDocumentGetRequest,
    ) -> PublicIndexedDocument: ...
    async def wait_for_document(
        self,
        request: LibraryDocumentGetRequest,
        *,
        timeout: float = 360.0,
    ) -> PublicIndexedDocument: ...
    async def delete_document(
        self,
        request: LibraryDocumentDeleteRequest,
    ) -> None: ...

The request and response models in the signatures above are importable from dualeai. Ignore the receipt’s location field, and use its library_id and document_id to build every document request. For processing failures, wait_for_document returns a failed document instead of raising an exception. Its document.failure contains RFC 9457 details, or None if the cause is unavailable. Document ingestion failures lists the known causes.

Limits and ceilings

Library limits covers file size, path, tags, access-list limits, upload and deletion windows, and the model’s Library Tool ceilings. The table above covers method arguments.

Capabilities the client does not wrap

Use the Dashboard for these capabilities; the core client provides no methods for them:

See Manage libraries for the supported workflow and SDK errors for DualeAIAuthError and DualeAIConnectionError.

Library actions

library:upload applies to the Tenant; the other actions apply to the Library’s path, library:<path>. Access to one path does not imply access to others. Use these identifiers when requesting a grant; the Dashboard uses plain-language labels.

An upload requires both Tenant-scoped library:upload and library:write on the Library’s path. Creation requires only library:write on the requested path. See LibrariesClient for path reuse and list resolution.

Access and isolation states who grants each action, how deny takes precedence over allow, and which boundaries a Grant does not draw.

BackpressureConfig

Pass backpressure_config= to DualeAISDK to tune cached-activity capacity and the process-local task circuit breaker.

from dualeai.backpressure import BackpressureConfig

config = BackpressureConfig(
    max_concurrent_activities=100,
    max_pending_activities=500,
    failure_threshold=10,
    recovery_timeout_seconds=30.0,
)

Task submission does not use the activity scheduler. A Task submission raises RuntimeError only while the circuit breaker is open.

MockSDK

MockSDK includes keyed task responses and an in-memory sdk.libraries client:

import asyncio

from dualeai import LibraryCreateRequest
from dualeai.testing import MockSDK


async def main() -> None:
    async with MockSDK() as mock:
        mock.set_mock_response("Summarize invoice", {"result": "data"})
        result = await mock.mock_ask("Summarize invoice")
        print(result)

        library = await mock.libraries.create(LibraryCreateRequest(path="tests/invoices"))
        print(library.id)


asyncio.run(main())

mock.libraries runs the core Library workflow in memory; wait_for_document() completes queued documents. mock_ask() looks up a saved response without running ask(), streaming, Tool dispatch, or terminal errors. To exercise ask() without a live connection, inject a mock through transport=.

The in-memory Library client does not enforce permissions or simulate file storage, asynchronous processing, or retention. Its delete methods remove records immediately. Use it to test application flow, then verify those service boundaries separately.

Versioning and status

The SDK is in Public Preview. A feature page states its own release stage when that stage differs from the rest of the SDK. To report the version installed in your Python environment, run:

python -m pip show dualeai