--- title: "Application-facing SDK API reference" description: "The Duale AI SDK application API defines task submission, conversation continuation, LibrariesClient, attachment, tool, and configuration interfaces." lang: en status: public-preview lastUpdated: 2026-09-04 url: https://duale.ai/en/docs/sdk/reference --- ## AI-generated summary Application-facing Python SDK reference covering task submission, continuations, AgentResponse, routing, configuration, decorators, attachments, and LibrariesClient operations. - Covers ask, continue_conversation, and AgentResponse methods including model, stream, next, stop, and cache_hit. - Documents DualeConfig environment variables, RoutingPolicy fields, ResponseFormat options, and SkillEnum labels. - Describes @tool, @activity, and @agent decorators with cache TTL, retry, and JSON-serializability requirements. - Details LibrariesClient methods for creating, listing, uploading, and deleting Library documents with tenant_id required. - stop_task returns on acceptance; prior usage is billed; only the submitting agent can stop it. Summaries were generated by AI. Generative AI is experimental. --- 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. ```python 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 ``` | Parameter | Meaning | | ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `action` | The task instruction. Required, non-empty, and at most 100,000 characters. | | `res` | A Pydantic model (or basic type) to validate the result against. Sets the return type of `response.model()`. | | `routing` | A `RoutingPolicy` of model-selection preferences. When present, it replaces the policy that `skills` would derive. | | `skills` | When `routing` is absent, shorthand for its soft `required_skills` preference. It also labels the task type for observability. | | `response_format` | A `ResponseFormat` that constrains the output shape. | | `streaming` | When `True`, `response.stream()` yields content deltas as they arrive. | | `deadline` | A timezone-aware absolute `datetime`. Defaults to 30 minutes from submission. A naive value currently raises `TypeError` before submission. | | `attachments` | Prepared attachment references to include with the task. Upload the files first with `sdk.upload_attachments`; see [Attach documents](https://duale.ai/en/docs/sdk/attachments.md). | | `request_id` | A caller-supplied task identifier: 10–128 ASCII letters, digits, `_`, or `-`. Defaults to a generated identifier. | | `sdk` | The `DualeSDK` instance. Required and keyword-only. | ## continue\_conversation Submit a non-streaming follow-up linked to an accepted conversation. ```python async def continue_conversation( response: AgentResponse[T], message: str, *, deadline: datetime | None = None, response_format: ResponseFormat | None = None, ) -> AgentResponse[T] ``` | Parameter | Meaning | | ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `response` | The previous `AgentResponse`. The function waits for it to finish successfully, then uses its task identifier and SDK instance. | | `message` | The next user message. Required and at most 100,000 characters; it cannot contain NUL. Empty and whitespace-only strings are accepted. | | `deadline` | An aware absolute `datetime`. A naive value raises `ValueError`. Defaults to 30 minutes from submission and cannot extend the current accepted conversation deadline. | | `response_format` | The output shape for this turn. Omit it to use the normal unstructured response. | 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](https://duale.ai/en/docs/sdk/concepts.md). `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()`. | Member | Behavior | | -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `task_id` | The durable task identifier. Available as soon as the call returns. | | `await model()` | Await the terminal result, validated against the response model. Raises `DualeError` on a platform failure, `TaskStoppedError` when the task was stopped, `ValidationError` on a schema mismatch. | | `stream()` | Async iterator of content deltas and content-reset events. Yields nothing when `streaming` is `False`; continuations are always non-streaming. | | `await next(message, ...)` | Wait for this task to succeed, then submit a child. `res=` optionally selects and validates a new result type for that child. | | `await stop(reason)` | Stop this task and everything under it. Returns as soon as the platform accepts; does not wait for the task to end. | | `await cache_hit()` | Whether the result was served from cache, or `None` if unknown. | | `await llm_metrics()` | The full result-metrics object, or `None`. | ## stop\_task Stop a running task and every task started under it. ```python async def stop_task(self, task_id: str, reason: str) -> TaskStopAccepted ``` | Parameter | Meaning | | --------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `task_id` | The task to stop. | | `reason` | Why it is being stopped. Required, at most 500 characters, and not only spaces. `TaskStopAccepted` comes from `duale.models.task_stop`. | 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. ```python 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. ```python 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. ```python 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. | Variable | Field | Default | Notes | | ------------------------------- | ------------------------ | ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `DUALE_TOKEN` | `token` | required | 10–256 characters, starts with `duale_`. | | `DUALE_ENDPOINT` | `endpoint` | `https://api.duale.ai` | Must start with `http://` or `https://`. Use `https://` outside local development: Library calls send the API token as a bearer header, so an `http://` endpoint puts the token on the wire in clear text. | | `DUALE_AGENT_ID` | `agent_id` | none | Required for `@tool`, `serve()`, and task-scoped attachment uploads. Starts with a letter; letters, digits, hyphen, underscore; 3–50 characters. | | `DUALE_TENANT_ID` | `tenant_id` | none | Required for Library management and attachment uploads. | | `DUALE_REDIS_URL` | `redis_url` | `redis://localhost:8010` | Optional cache; falls back to SQLite. Must start with `redis://`. | | `DUALE_SQLITE_PATH` | `sqlite_path` | none | SQLite cache file path. | | `DUALE_DEBUG` | `debug` | `false` | Verbose logging. | | `DUALE_OBSERVABILITY__ENDPOINT` | `observability.endpoint` | `http://localhost:4318` | OTLP over HTTP base endpoint. A separate outbound destination from `DUALE_ENDPOINT`: allow it in your egress policy when you export telemetry off the host. | | `DUALE_OBSERVABILITY__TOKEN` | `observability.token` | none | Enables telemetry export. Use a telemetry token, never your `duale_` API token. | ## 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. | Field | Range | Omitted behavior | Meaning | | ----------------------- | ------------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | | `target_accuracy` | 0.0–1.0 or `None` | No emphasis | `0` or `None` adds no capability emphasis; higher values favor the highest available capability. | | `target_permissiveness` | 0.0–1.0 or `None` | No preference | `0` favors qualified answers, `0.5` balances, and `1` favors direct answers to sensitive but allowed requests. Not a safety control. | | `cost_sensitivity` | 0.0–1.0 or `None` | No base-tariff preference | `0` or `None` adds no base-tariff preference; higher values favor cheaper routes. Cache savings can affect every value. Not a budget. | | `speed_preference` | 0.0–1.0 or `None` | Normal balance | `0` disables measured-latency preference, `None` or `0.5` applies the normal balance, and `1` is strongest. | | `priority_level` | -10–10 | Normal balance | Adjusts measured-latency preference. It does not set queue priority. | | `required_skills` | list of `SkillEnum` | Empty | Prefer models with every listed skill; relax to the best available route when unmet. | | `preferred_skills` | list of `SkillEnum` | Empty | Improve preference without excluding other routes. | 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: ```python 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](https://duale.ai/en/legal/cgu.md) 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: ```python @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](https://duale.ai/en/docs/sdk/tools.md). `@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](https://duale.ai/en/docs/sdk/tools.md). ## 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](https://duale.ai/en/docs/sdk/concepts.md). ## 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](https://duale.ai/en/docs/sdk/attachments.md). ## 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. ```python 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: ... ``` | Method | Contract | | --------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `create(request)` | **Contract:** Takes `LibraryCreateRequest`, which carries only `path`; the server assigns the Library id. Returns an active Library at the path when the caller can change it. Otherwise creates a new Library and grants the caller access. Paths are non-unique; inaccessible and deleted Libraries do not reserve them. **Failure:** Returns `409` or `503` when the platform cannot finish creating the Library. A partly created Library does not appear in `list()`, and the platform refuses reads against it. **Recovery:** Repeat `create(request)` with the same path and caller identity. It finishes the Library that the first attempt started instead of creating a second one. | | `list()` | Returns `LibraryListResponse`; its `libraries` field contains the live Libraries visible to the caller. It fails with `BusinessError` rather than truncating once the caller can reach more than 5,000 resources. | | `get(request)` | Takes `LibraryGetRequest`. Returns an active Library's stable id, current path and tags, revision actor and time, and creation time. A deleted Library returns `410`. | | `update(request)` | Takes `LibraryUpdateRequest` with a nested `LibraryPatchRequest`. The patch changes `path`, replaces the complete `tags` map, or both. `tags={}` clears all tags. Changing the path does not change the Library id. | | `delete(request)` | **Contract:** Takes `LibraryDeleteRequest`. Stops normal reads and writes, makes every document in the Library unrecoverable, and returns the same result for a live, deleted, or unknown Library. The record and grants become eligible for asynchronous removal after 24 hours. **Failure:** A `503` means the Library is temporarily unavailable for deletion; it stays in place and no data is lost. **Recovery:** Retry later. Report the failure with the `request_id` if it persists. | | `upload(library_id, attachments)` | **Contract:** Uploads prepared files concurrently and returns queued receipts by attachment key. The batch is not transactional. **Recovery:** On `LibraryUploadError`, inspect `completed_receipts`. After a platform error, list the documents to reconcile prior successes. | | `list_documents(request)` | Takes `LibraryDocumentListRequest` with required `limit`, 1 to 500, and nullable `cursor`. Returns one `LibraryDocumentPage`. Documents in the Recently deleted window are excluded. | | `get_document(request)` | Takes `LibraryDocumentGetRequest` and reads one document state without polling. | | `wait_for_document(request)` | Takes `LibraryDocumentGetRequest`, polls every five seconds with `GET`, and returns on `ready` or `failed`. The positive `timeout` bounds the whole polling operation and defaults to 360 seconds. | | `delete_document(request)` | Takes `LibraryDocumentDeleteRequest` and moves one document into the 30-day Recently deleted window. Repeated calls succeed. The core Python client does not expose restore. | 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](https://duale.ai/en/docs/sdk/errors.md) lists the causes. ### Limits and ceilings The method table above states the arguments your code passes. [Library limits](https://duale.ai/en/docs/libraries/limits.md) 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. | Capability | What it does | | ---------------- | --------------------------------------------------------------------------------- | | Grants | Add and revoke access to one Library. Needs `library:admin` on it. | | Document preview | Read the opening of a document's extracted text. The only content read-back path. | | Document tags | Change a document's tags after upload. `upload()` sets none. | | Recently deleted | List the tenant's deleted documents. | | Restore | Return a deleted document to its Library. Needs `library:write`, within 30 days. | 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](https://duale.ai/en/docs/sdk/manage-libraries.md) for the supported workflow and [Errors and reliability](https://duale.ai/en/docs/sdk/errors.md) 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. | Action | Scope | Operations that check it | | ---------------- | ------- | ---------------------------------------------------------------------------------------- | | `library:upload` | Tenant | `create()`, and starting a document upload | | `library:read` | Library | `get()`, `list_documents()`, `get_document()`, `wait_for_document()` | | `library:write` | Library | `create()` when it adopts an existing path, `update()`, `upload()`, tag changes, restore | | `library:delete` | Library | `delete()`, `delete_document()` | | `library:admin` | Library | Reading and changing that Library's grants | 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](https://duale.ai/en/docs/libraries/access.md) 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. ```python 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: ```python 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. ## Related content - [Python SDK for bounded agent work with typed results](https://duale.ai/en/docs/sdk.md) - [Declare and serve Python tools with the SDK](https://duale.ai/en/docs/sdk/tools.md) - [SDK task lifecycle, routing, and streaming](https://duale.ai/en/docs/sdk/concepts.md) - [Integrate model routing in an application](https://duale.ai/en/docs/model-routing/application-integration.md) - [Attach files to a task using the SDK](https://duale.ai/en/docs/sdk/attachments.md) - [Manage Libraries and documents with the SDK](https://duale.ai/en/docs/sdk/manage-libraries.md) --- ## Sitemap See the full [Markdown sitemap](https://duale.ai/sitemap.md) for all pages.