SDK task lifecycle, routing, and streaming

The SDK is a thin client to a hosted runtime. Your process submits a bounded task; the platform uses eligible model routes and returns a terminal result. It can try another eligible route before that result. Your code never calls a model provider directly.

The task request path

Your process opens one HTTP connection per task and submits the task on it. The platform selects a model route from the pool your tenant enabled and runs the task. It streams content deltas and the terminal result back over server-sent events on that same connection.

Your process opens every connection. Duale AI opens none to it, so the SDK host needs no inbound port and no public address: allow outbound HTTPS and it works from a private network segment. That covers the SDK host only. A model provider you configure at your own address is one the platform reaches, and Plan the deployment states the network path you supply for it.

The platform resolves your tenant and agent identity from the token on each request. Identifiers your code puts in a payload are never the source of truth for access.

Library management uses /libraries on the same origin as your endpoint, authenticated with your API token. File uploads obtain presigned URLs and send file parts directly to object storage. Library requests do not use the task event stream.

Tasks and results

ask() submits one task and returns an AgentResponse immediately, before the work finishes. Two accessors read the outcome:

• await response.model() blocks until the task reaches a terminal state and returns the result, validated against the response model you passed. A platform failure raises DualeError; a schema mismatch raises ValidationError.
• response.task_id is the durable identifier for the task. It is available as soon as ask() returns, so you can log it before the result arrives.

The task carries an absolute deadline. Without an explicit deadline, the SDK sets one 1800 seconds (30 minutes) ahead. A task that passes its deadline ends as a failure, not a partial result.

Cancelling the local wait does not stop the task. Cancelling response.task stops your process from awaiting the stream, but the platform keeps running the task.

To stop the work on the platform, call await response.stop(reason). The task then ends with a stopped result, and await response.model() raises TaskStoppedError. Stopping does not undo a call already sent to a model provider or a tool, and it does not refund what the task already consumed.

Routing policy

You do not name a model. You pass a RoutingPolicy of soft preferences. The runtime uses it to rank eligible routes and can try more than one route before it returns the terminal result:

from duale import RoutingPolicy

policy = RoutingPolicy(
    target_accuracy=0.9,       # favor higher-capability models
    cost_sensitivity=0.3,      # mild preference for cheaper routes
    speed_preference=0.5,      # normal latency balance
    required_skills=["analysis"],
)

Every field is optional and expresses a preference, not a hard filter. required_skills prefers evidence for every listed skill and relaxes when the active pool cannot satisfy the full request. target_permissiveness tunes how often a model withholds or qualifies sensitive-but-allowed answers; it is a routing preference, not a safety guardrail.

Routing contract and limits lists every field, its range, and what routing does not guarantee.

Streaming

Pass streaming=True to observe content as the model produces it. response.stream() yields BridgeContentDeltaResponse events. Read each event’s .delta string. A BridgeContentResetResponse tells you to discard everything rendered so far because replacement content follows:

import asyncio

from duale import BridgeContentResetResponse, ask, create_sdk


async def main() -> None:
    async with create_sdk() as sdk:
        response = await ask("Write a short poem", streaming=True, sdk=sdk)

        buffer = []
        async for event in response.stream():
            if isinstance(event, BridgeContentResetResponse):
                buffer.clear()
                continue
            buffer.append(event.delta)

        poem = await response.model()
        print(poem)


asyncio.run(main())

Streamed deltas are for display only. await response.model() returns the source of truth. Always take the final result from it, not from the accumulated deltas. Replace the displayed text when the stream ends. Never store, index, or forward the accumulated preview as the answer. Only the terminal result carries the content mark.

Tools run in your process

A task can call functions you declare with @tool. Those functions run in your own process, not on the platform. You publish them with sdk.serve(), which holds them available through a heartbeat; each call then reaches your process on the event stream of the task using the tool. Authoring tools owns delivery replay, idempotency, retries, and the full serving flow.

What the platform manages, and what you own

Use this boundary to decide where your application needs a control:

When you enable an image-capable model, images inside an indexed document reach it as input, not only their extracted text. See Images in a document.

Multi-turn continuation

response.next() submits a follow-up after its parent reaches a successful terminal state. A parent error or local cancellation stops the call before the SDK submits a child. continue_conversation(response, message) provides the same operation as a standalone function. Both APIs require the previous response object. A task identifier alone cannot continue a conversation.

Each follow-up gets a client-generated child task identifier. The platform links that child to the accepted conversation head. It preserves prior user and assistant text, accepted tool calls and results, the initial routing policy, and original attachment references. Two children can continue the same parent. Both inherit the parent context, but neither sees the other child’s turns. The child stays in the parent’s tenant and agent scope.

Continuation responses are non-streaming, so continuation methods expose no streaming option. A continuation can set its own deadline and response_format; the platform caps that deadline at the parent’s current accepted conversation deadline. The returned response’s task_id is the new child task identifier.

Long-running work

Keep related steps in one conversation when later work depends on earlier conclusions.

During a long run, the model does not receive the whole conversation. Well before the selected route’s context window fills, the platform includes less of the older history in later model calls.

These inputs do not have the same lifetime:

Put every rule the agent must keep in the root task instruction passed to ask(). A follow-up cannot amend that instruction or grant system authority.

Reduction does not delete the stored history. After reduction, the model can call list_context_history to list references for recent items or for a selected time range. It can call recall_context to read one original, optionally by line range. Agent harness explains when the platform adds these tools. Authoring tools lists their reserved names.

What you control

Each route has a context_window. Your deployment can also set its effective_context_window. SDK tasks cannot change these fields.

context_window caps the total tokens for input and output. effective_context_window records a measured quality limit. Set it to the largest context at which the model maintains quality. When set, the platform uses it to judge context fit and decide how much history to keep. Leave it unset without a reliable measurement.

For each task, you control:

• Tool results. Return only what later steps need. Shorter results can remain in the active context longer.
• Conversation boundaries. Call ask() again when work no longer needs prior conclusions. The new conversation has no prior history.
• The deadline. Choose an absolute deadline. It can be hours or days ahead. An interrupted task cannot resume.

One tool call can make retries + 1 attempts. Budget one timeout per attempt plus time for backoff. If less than timeout × (retries + 1) remains, later retries can fail to start. Scope and limits explains what an interruption does.

When this fits

One conversation fits research that accumulates findings, a queue processed item by item, or a revision that builds on prior conclusions. It fits less well when a later step must quote early content exactly. Have the tool return the value again, or use separate ask() calls.

A long run in one process

Use this example when one process owns the task from submission to its terminal result:

import asyncio
from datetime import UTC, datetime, timedelta

from duale import DualeError, TaskStoppedError, ask, create_sdk


async def main() -> None:
    async with create_sdk() as sdk:
        response = await ask(
            "Audit the invoices in the finance Library. Never approve an invoice above 10,000 EUR; "
            "flag it for a human instead. Report each decision with its invoice id.",
            deadline=datetime.now(UTC) + timedelta(hours=4),
            sdk=sdk,
        )

        try:
            print(await response.model())
            follow_up = await response.next("Summarize the flagged invoices for the approver.")
            print(await follow_up.model())
        except TaskStoppedError as error:
            print(f"stopped on purpose: {error}")
        except DualeError as error:
            print(f"the run failed: {error}")


asyncio.run(main())

Keep response.task_id to reconcile the run later. The Dashboard shows the provider and model calls for the task and the input tokens on each call. The API does not export that view.

Use one conversation for one chain of dependent conclusions. Call ask() again when the next piece of work does not need that history.