Application-facing SDK API reference

This page covers the application-facing task, continuation, LibrariesClient, attachment, tool, and configuration APIs. Common convenience imports come from duale; specialized classes use the module paths shown below, and testing helpers live in duale.testing. Bridge wire models remain importable from duale.models.bridge for callers of lower-level SDK methods. Other generated and transport modules are not covered by this application reference unless named below. The distribution is published as dualeai; the import package is duale. Signatures show the application contract; keyword-only arguments follow the * marker.

ask

Submit one task and return an AgentResponse immediately.

async def ask(
    action: str,
    skills: list[SkillEnum] | None = None,
    res: type | None = None,
    response_format: ResponseFormat | None = None,
    routing: RoutingPolicy | None = None,
    *,
    streaming: bool = False,
    deadline: datetime | None = None,
    attachments: list[PreparedAttachment] | None = None,
    request_id: str | None = None,
    sdk: DualeSDK,
) -> AgentResponse

continue_conversation

Submit a non-streaming follow-up linked to an accepted conversation.

async def continue_conversation(
    response: AgentResponse[T],
    message: str,
    *,
    deadline: datetime | None = None,
    response_format: ResponseFormat | None = None,
) -> AgentResponse[T]

The SDK generates a child task identifier before it submits the request. The returned AgentResponse.task_id is that child identifier, while response.task_id remains its public parent task identifier. The platform preserves accepted text, tool turns, routing context, and original attachment references. See How the SDK works.

response.next(message, *, res=None, deadline=None, response_format=None) uses the same continuation request after it waits for the response to finish successfully. res= selects the child’s runtime validation model and, when response_format= is absent, derives the child’s JSON Schema. The current type annotation retains the parent’s AgentResponse[T] parameter, so a type checker does not infer a different child type from res=. If the parent fails or the local task is cancelled, .next() propagates that outcome and does not submit a child.

AgentResponse

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

stop_task

Stop a running task and every task started under it.

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

The call returns as soon as the platform accepts the request; TaskStopAccepted carries the task and the moment of acceptance. A task you own and that still runs then ends, and awaiting its model() raises TaskStoppedError with the reason you supplied. stream() yields no stop event: it ends as it ends for any other outcome, so read the outcome from model().

A task that answered before the stop landed keeps its own result, and model() returns it—one task never produces two outcomes. Repeating the same stop is safe and produces the same single stopped result.

from duale 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)

response.stop(reason) is the shortcut for a task you already hold, in the same family as response.next(). sdk.stop_task(task_id, reason) takes the identifier directly, for a task this process did not submit. Only the agent that submitted a task can stop it. A stop for another agent’s task is accepted and then ignored. Confirm a stop by reading the task outcome, not the acceptance.

Stopping does not undo work already sent to a model provider or a tool. It also does not refund usage recorded before the stop. That usage is billed as it is for a task that finishes on its own. Work that the stop prevents never starts, so it costs nothing.

DualeSDK

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

class DualeSDK:
    def __init__(
        self,
        config: DualeConfig | 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 DualeConfig() built from the environment. max_jobs sets the default concurrency for cached activities and registered tools. job_timeout bounds cached activities. 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 them immediately. auto_start defaults to True on the constructor. The create_sdk() helper and MockSDK default it to False.

For a task-only client, use async with create_sdk() as sdk:. serve() publishes the registered tools for the configured agent_id and blocks. It runs heartbeats until shutdown. stop_event= provides cooperative shutdown for an embedding application.

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

create_sdk

create_sdk(**kwargs) builds a DualeSDK from keyword arguments, falling back to DUALE_ environment variables, and returns it with auto_start=False. It is the ergonomic constructor for the common environment-configured case.

from duale 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.

DualeConfig and environment variables

DualeConfig reads DUALE_ environment variables. Nested observability keys use a double underscore.

RoutingPolicy

Model-selection preferences. Every field is optional and every field is a soft preference. required_skills asks routing to prefer models with evidence for every listed skill, but it still falls back when the full request cannot be satisfied.

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

ResponseFormat

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

from duale.models.response_format import JsonSchemaResponseFormat

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

SkillEnum

general, reasoning, analysis, math, code, instruction_following, agentic, long_context, finance, cyber_defensive, cyber_offensive, medical, legal.

A skill label states what a model scored on public, reproducible benchmarks. It does not state a use Duale AI offers or recommends. The prohibited uses listed in the terms of use prevail over any label.

required_skills and preferred_skills are routing weights, not guardrails. They raise the chance of selecting a model that carries the skill; they never forbid a selection. When the pool cannot meet the full skill request, routing uses the best available eligible route.

Decorators: @tool, @activity, and legacy @agent

Use these signatures to declare tools, cached activities, and the legacy agent decorator:

@tool(*, sdk: DualeSDK, 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: DualeSDK)

@agent(name: str, *, sdk: DualeSDK)

@tool registers a function the platform can call. See Authoring tools. @activity caches an async function’s result for cache_ttl, which defaults to one hour. It retries the function up to max_retries extra times.

Its cache key comes from the function name and the repr() of its arguments. The result must be JSON-serializable. @agent records legacy local metadata only. It is not part of the hosted-tool lifecycle manifest.

Tool context fields

Inside a running tool, current_tool_context() returns a ToolContext with tool_call_id, attempt, task_id, and deadline_at (plus remaining_seconds() and is_expiring()), or None outside a tool call. See Authoring tools.

Streaming events

response.stream() yields BridgeContentDeltaResponse (a .delta string) and BridgeContentResetResponse (discard content rendered so far). Take the final result from await response.model(), not the accumulated deltas. See How the SDK works.

Attachments

prepare_attachments() returns PreparedAttachment references. await sdk.upload_attachments(task_id, attachments, agent_id=...) uploads each file to the task’s call-scoped Library and returns dict[str, LibraryDocumentCreateResponse], keyed by PreparedAttachment.key. Pass the configured identity with agent_id=sdk.agent_id. Omit the argument only when the SDK has exactly one agent registered through the legacy @agent decorator. See Attach documents.

LibrariesClient

sdk.libraries manages persistent Library metadata and documents. Every operation requires a configured tenant_id, either through DualeConfig(tenant_id=...) or DUALE_TENANT_ID. The server resolves the caller from DUALE_TOKEN and applies its Library grants.

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 duale. Ignore the receipt’s location field, and use its library_id and document_id to build every document request. A failed document is a normal terminal return from wait_for_document; document.failure carries RFC 9457 details, and Errors and reliability lists the causes.

Limits and ceilings

The method table above states the arguments your code passes. Library limits states every other number. These include the file size, the path and tags limits, and the 5,000-resource ceiling on list(). The page also states the upload and deletion windows and the ceilings for the model’s document tools.

Capabilities the client does not wrap

The platform carries five Library capabilities the core client does not wrap. Use the Dashboard for them.

Revision history has no client. GET {DUALE_ENDPOINT}/libraries/v1/tenants/{tenant_id}/{library_id}/revisions returns one Library’s past paths and tags, newest first, and checks library:read. Call that route directly, because neither the core client nor the Dashboard reads it; get() and update() already return the current revision.

See Manage libraries for the supported workflow and Errors and reliability for DualeAuthError and DualeConnectionError.

Library actions

Five actions decide which Library operations a caller can perform. Use this table to identify the required action and its scope before you request a grant.

The server checks library:upload against the tenant. It checks the other four against the Library the request names, so a subject can hold them on one Library without holding them tenant-wide. The Dashboard shows each action under a plain-language label rather than the identifier below.

Two operations depart from that table:

• create() checks twice. First library:upload on the tenant, then library:write on each Library already at the path. It adopts the first that passes and creates a new Library when none does.
• list() checks nothing per Library. It returns the Libraries the caller can reach with library:read, and raises when that reach exceeds 5,000 Libraries.

Every upload also checks two actions, one at each scope.

Access and isolation states who grants each action, how a refusal behaves, and which boundaries a grant does not draw.

BackpressureConfig

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

from duale.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. ask() and submit_task() raise RuntimeError only while the task dependency circuit is open.

MockSDK

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

from duale import LibraryCreateRequest
from duale.testing import MockSDK

mock = MockSDK()
mock.set_mock_response("Summarize invoice", {"result": "data"})
result = await mock.mock_ask("Summarize invoice")

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

mock.libraries supports the documented core Library workflow in memory and completes queued documents when you call wait_for_document(). mock_ask() is a keyed lookup, not the real ask() path; it does not simulate streaming, tool dispatch, or terminal errors. To drive ask() end to end without a live platform connection, inject a mock transport through the transport= constructor argument.

Versioning and status

The package version is 0.1.0 and the SDK is in Public Preview. A feature page states its own release stage when that stage differs from the rest of the SDK.