Integrate model routing in an application

Send business intent and routing preferences through the same SDK contract for every qualified route that supports the functions your application uses. Keep provider-specific selection and credentials out of application code.

Before you add routing preferences

Confirm these application and operator inputs before you add a soft preference:

• A provisioned API token and the complete Duale AI task API base URL for the target deployment, including any gateway path prefix. The SDK appends the concrete task-resource path. The token selects the task’s tenant and agent scope.
• DUALE_TENANT_ID and Library grants for Libraries; DUALE_AGENT_ID and lifecycle permission for hosted tools; both identifiers and Library access for attachments.
• A model pool approved and provisioned for the workload.
• A typed response model and measurable business acceptance criteria.
• A defined failure path when no eligible route returns an acceptable result before the deadline.

Stop when any boundary is missing. A routing policy cannot create it.

Set a contextual policy

Pass RoutingPolicy with the task. The values express relative preferences inside the active pool:

import asyncio
from datetime import UTC, datetime, timedelta
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

from duale import RoutingPolicy, SkillEnum, ask, create_sdk


class ReviewResult(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    decision: Literal["manual_review", "reject"]
    reason: str = Field(
        min_length=1,
        max_length=1000,
        description="Why you reached this decision, in at most 1000 characters.",
    )


async def main() -> None:
    policy = RoutingPolicy(
        target_accuracy=0.85,
        cost_sensitivity=0.25,
        speed_preference=0.8,
        required_skills=[SkillEnum.analysis],
    )

    async with create_sdk() as sdk:
        response = await ask(
            action="Review this case and propose the next action.",
            res=ReviewResult,
            routing=policy,
            deadline=datetime.now(UTC) + timedelta(minutes=2),
            sdk=sdk,
        )
        result = await response.model()
        if result.decision == "manual_review":
            print(response.task_id, "send to the independent review queue")
        else:
            print(response.task_id, "reject without an external side effect")


asyncio.run(main())

Restate every limit in its field description. A provider that enforces your schema enforces its structure, not its values. It does not enforce a length, range, or pattern limit. The model does not see that limit unless the description states it. Without it, the answer fails validation only after you have paid for it. See Provider capability and compatibility.

Use each field for one clear intent:

Do not derive provider names, model versions, or allowed uses from these values. The routing contract defines the exact value and None behavior.

A non-empty skills= list creates a policy that contains only required_skills; it does not combine with the service default policy. None or an empty list leaves the policy absent. When you pass both arguments, the explicit RoutingPolicy wins; the SDK does not merge them.

Make the workflow safe to change

The model pool can change without an application deployment. Protect the process at its stable boundaries:

Protect business state

Apply this sequence so route changes and task retries do not become duplicate business changes:

1. Create a durable business-item key and record the item before task submission.
2. Validate the terminal result with a strict schema and deterministic domain rules. Model-reported confidence is not an approval control.
3. Move a high-impact proposal to a human-review state, then execute an accepted action as a separate idempotent step. A tool can instead verify a durable pre-approval at its boundary. Do not block inside a replayable tool while a person decides.
4. Restrict the agent’s registered tools and make every side effect idempotent. A request has no tool allowlist, and tool delivery can repeat during recovery.
5. Store the non-secret deployment identifier, task tenant and agent labels supplied by the operator, task_id, and business-item key with the business record. Never store DUALE_TOKEN or provider credentials there. Scope the task identifier to its deployment and tenant in your records; do not use it as the durable business-item key.
6. Keep authoritative business state outside the conversation, and apply one atomic state transition before any external action. This prevents a retry, callback, restart, or repeated approval from applying the result twice.
7. Set a deadline and define paths for stopped work, validation failure, submission failure, retryable platform problems, owner-action problems, and unacceptable results.

These boundaries let the application repeat or stop task work without repeating an accepted business effect.

Recover without repeating work

Terminal task failures and some earlier HTTP failures attach problem details to a DualeError or subclass. Check exc.problem_details before reading its error code or retry fields. TaskStoppedError has no problem details, and result validation raises ValidationError. See Handle SDK errors.

The SDK and task runtime already perform some retries. A new task identifier creates new work; an exact resubmission with the same task identifier reuses the admitted task. Respect retry_after_seconds, require business idempotency, limit attempts, and do not wrap ask() and model() in one generic retry loop.

For a batch process, use explicit states such as queued, submitted(task_id), retry_wait, accepted, review, and terminal_failed. Persist request_id before submission; it becomes the task identifier. Reuse it only with unchanged input after an ambiguous submission failure. It is not a business-side-effect key or a general restart interface. The application owns submission concurrency, budget admission, and recovery. SDK max_jobs limits cached activities, not submissions.

Separate cache evidence from provider evidence

An identical request can return an exact response-cache hit without a new provider call. await response.cache_hit() returning False proves that the terminal result was not an exact response-cache hit; None does not prove a provider call. The public request has no cache-bypass option. A freshness-sensitive workflow therefore needs a qualified path that can produce and verify a cache miss, or it must stop safely.

Before deployment, exercise each business state, recovery path, and cache outcome in the tests below.

Test outcomes, not route identity

Use three test levels:

• deterministic application tests for business states, validation, idempotency, review, and recovery;
• SDK protocol tests for submission, streaming reset, stop, error classes, and tool replay;
• live route evaluations for provider behavior, deadlines, pool exhaustion, cache miss, and route changes.

Build the evaluation set from representative business cases, difficult edge cases, and known failure modes. Include:

• schema, domain, and complete-workflow acceptance;
• tool permissions, side effects, and replay;
• latency, deadlines, pool exhaustion, and verified cache misses;
• streaming reset and terminal replacement; and
• pool changes that preserve the SDK interface but alter business behavior.

The SDK MockSDK does not simulate streaming, tools, or terminal errors. Do not use it as proof for those paths.

Do not make application tests pass only when one provider produces a particular phrase. Keep provider-specific route qualification in Control model changes.

What a continuation keeps

A continuation keeps the root request’s policy value, SDK instance, token-derived scope, tenant, and deployment. It cannot add attachments, set a new policy, stream, or extend the accepted deadline. It does not freeze the model pool, so a later turn can use configuration that changed after the root task. Start a new root task when the attachment set, deployment, hard boundary, or routing policy must change.

Configure one deployment per environment

Each deployment needs its own provisioned pool, token, identifiers, grants, and SDK client. The handoff must supply the complete DUALE_ENDPOINT task API base URL, including any gateway prefix. Do not append /v1/tasks/{task_id}; the SDK adds it. Use HTTPS outside local development. A new task does not select a named pool or transfer conversation state.

Keep the token, tenant and agent identifiers, activity-cache location, and telemetry settings deployment-specific. When used, the same origin must expose /libraries, and the client must reach each presigned storage host. Apply the required access, encryption, retention, deletion, and backup rules to Redis or SQLite activity-cache data.

SDK-managed telemetry uses a process-global identity and exporter configuration. Use separate processes when the deployment, token-derived tenant identity, telemetry endpoint, or telemetry token differs, or use a qualified host-owned telemetry setup. The deployment handoff supplies the supported SDK and platform pair and their upgrade order.

Treat each deployment as independently provisioned and qualified. Use a new root task when work must cross that boundary.