Declare and serve Python tools with the SDK

Let a task call your own Python functions. You declare each function with @tool, then run sdk.serve() to publish them and keep them alive. The functions run in your process, and each call reaches them on the task’s own event stream, over a connection your process opened.

Before you serve a tool

Check these three conditions before registration:

• A DUALE_TOKEN and matching, centrally provisioned DUALE_AGENT_ID. The SDK never creates agents; serving tools without an agent identifier raises an error.
• Each @tool needs a sdk, a model-facing description, and a timeout.
• Tool inputs and outputs must be types the SDK can serialize. A Pydantic model is the clearest choice.

Meet every condition before you call serve(); the SDK rejects a missing value or unsupported type instead of choosing one for you.

Declare and serve a tool

This process declares one typed tool and publishes it. Submit the task that uses it from the same process, so that one process both publishes the tool and holds the task stream the call arrives on:

import asyncio
from datetime import timedelta

from pydantic import BaseModel, ConfigDict

from duale import DualeConfig, DualeSDK, current_tool_context, tool


class GateCommand(BaseModel):
    model_config = ConfigDict(extra="forbid")

    gate_id: str
    reason: str


class GateState(BaseModel):
    gate_id: str
    state: str


async def main() -> None:
    sdk = DualeSDK(config=DualeConfig())

    @tool(
        sdk=sdk,
        description="Close a named security gate and return the final gate state.",
        timeout=timedelta(seconds=10),
    )
    async def close_security_gate(command: GateCommand) -> GateState:
        context = current_tool_context()
        if context is not None and context.is_expiring():
            raise TimeoutError("not enough time left to close the gate safely")
        # Must be idempotent: the platform can redeliver this call after a restart.
        return GateState(gate_id=command.gate_id, state="closed")

    await sdk.serve()


asyncio.run(main())

sdk.serve() publishes the tools and runs until the host stops it or a lifecycle failure ends the service. close_security_gate stays a normal function—the decorator registers it without changing how it is called.

While serve() runs, the process publishes the tool set for agent_id and sends a heartbeat about every 60 seconds. Publishing is what makes the tool available to the model; the call itself arrives on the event stream of the task that is using it. One process must do both. A process that only publishes never receives a call, and it still brings the agent online, so a split across two processes fails quietly: the model selects the tool, no host answers, and the call runs out its deadline.

Set DUALE_DEBUG=true to check both halves: the registration and heartbeat log lines confirm the tool is published, and the task’s result confirms the function ran.

What the model sees

Before a call, the language model receives the tool name, its description=, and the parameter JSON Schema. That schema includes parameter names, types, constraints, and hints written as Annotated[T, Field(...)]. A plain Python docstring stays developer-facing and is not sent. After execution, the tool result or the configured or default error text returns to the model. To describe and constrain a parameter, annotate it:

from typing import Annotated

from pydantic import BaseModel, Field


class Rating(BaseModel):
    score: Annotated[int, Field(ge=1, le=10, description="Quality from 1 to 10")]

The schema the model sees is built from the same annotations the SDK validates against, so a constraint you annotate is both advertised and enforced. Constraints run when a call arrives.

Build parameters from Pydantic models. A nested class of another kind still validates, but it advertises no restriction to the model, so the model gets no hint about what it must not send.

Reject input you did not declare

The SDK rejects an unexpected top-level parameter. It does not reject one inside a model you pass as a parameter: that model keeps its own Pydantic configuration, and the default drops an unknown field silently. Your tool then runs on the fields it recognized and returns a normal result, so nothing reports the bad input.

Configuring the outer model closes only the outer level. Give every model in the tree the same strict base:

# Illustrative — inherit this from every model you accept, at every depth.
from pydantic import BaseModel, ConfigDict


class Strict(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ReviewOptions(Strict):
    include_history: bool


class ReviewRequest(Strict):
    document_id: str
    options: ReviewOptions

You cannot add that base to a model another library owns. When a parameter must carry a vendor model, wrap it: declare your own strict model with the fields you accept, and build the vendor object inside the function.

The serve lifecycle

sdk.serve() runs three phases against the platform:

1. Register. It publishes the full tool set for the configured agent_id.
2. Heartbeat. It sends a heartbeat about every 60 seconds, and republishes the tool set when it changes.
3. Deregister. On shutdown it makes a best-effort deregistration call, then closes its connections.

Every process serving one agent shares one lifecycle request budget: the platform counts each register, heartbeat, and deregister call per agent, not per process. Once replicas exhaust it, the platform starts refusing their heartbeats and serve() raises. Give a busy workload its own agent rather than stacking replicas on one.

Checked on 2026-09-03, an agent reaches that budget at around a dozen concurrent processes, and at far fewer while the platform’s shared counter is unreachable. Current behavior, not a commitment.

The lifecycle payload’s agent_id must match the token-bound identity. The platform hides tools for unknown, deleted, or unauthorized agents. Call sdk.serve(stop_event=event) when an embedding application needs cooperative shutdown. Have the host set that event or cancel the serve task when it handles SIGINT or SIGTERM; abrupt termination can skip best-effort deregistration. A registration or sustained heartbeat failure propagates from serve(), so the host must supervise it and apply its restart policy.

Limits

The platform validates the manifest at registration, before serve() starts:

• A tool timeout must be greater than 0 and at most 3600 seconds (one hour).
• One agent serves at most 64 tools.
• A tool name must match ^[a-zA-Z0-9_-]+$, be at most 64 characters, and must not be one of the reserved names exec_agent, finish, list_agents, list_context_history, or recall_context. A duplicate name is rejected.
• A description is at most 1024 characters.

A tool that violates a limit raises at registration time, so serve() never starts with an invalid manifest.

Idempotency and at-least-once delivery

Tool delivery is at-least-once. Within a single running process the SDK deduplicates a redelivered call by (task_id, tool_call_id) and returns the cached result. That cache is process-local: after a restart, the platform can redeliver a call and run your function again. A timeout does not show whether your code ran, so the side effect is uncertain.

Build that protection on three rules:

• Key on (task_id, tool_call_id), never on either half. The model provider issues the call identifier, and attempt restarts at 1 on every redelivery, so neither is unique alone.
• Give a repeatable effect its own discriminator in the tool input. Not even the pair is guaranteed unique per request, so a second legitimate request can be absorbed as a duplicate.
• Retain your duplicate store for the tool’s timeout, not for the redelivery window. The deadline is replayed unchanged, and the SDK refuses an expired call before it invokes your function.

Make side-effecting tools idempotent or safe to retry. Secure integration owns the durable claim procedure and both ways to handle a repeatable effect.

Retries

Retries are opt-in per tool and default to 0:

# Illustrative — extends the example above; sdk, GateCommand, and GateState are reused.
@tool(sdk=sdk, description="...", timeout=timedelta(seconds=10), retries=2)
async def do_work(command: GateCommand) -> GateState: ...

A retry runs the function again on failure, up to retries extra times, bounded by both the tool timeout and the task deadline, with jittered backoff. Enable it only for idempotent or retry-safe tools—it is a second at-least-once source on top of delivery replay. If the deadline is too short to fit the configured retries, some will not fire and the SDK logs a warning.

Use the current tool context

Inside a running tool, current_tool_context() returns a ToolContext with:

• tool_call_id—the call identifier, constant across retries of the same call;
• attempt—the 1-based attempt number, counted within one delivery;
• task_id—the owning task;
• deadline_at—when the call must finish, with remaining_seconds() and is_expiring() helpers.

Outside a tool call it returns None, so a direct unit-test call sees None.

Use task_id and tool_call_id together as the idempotency key, never either one alone, and never attempt. Idempotency and at-least-once delivery states what that key does and does not separate.

Test a tool without the platform

A @tool-decorated function stays directly callable, so you can test its logic without the platform:

import asyncio
from datetime import timedelta

from pydantic import BaseModel, ConfigDict

from duale import create_sdk, tool


class GateCommand(BaseModel):
    model_config = ConfigDict(extra="forbid")

    gate_id: str
    reason: str


class GateState(BaseModel):
    gate_id: str
    state: str


async def main() -> None:
    sdk = create_sdk(token="duale_local_test_token", agent_id="agent_local_test")

    @tool(sdk=sdk, description="Close a named security gate.", timeout=timedelta(seconds=10))
    async def close_security_gate(command: GateCommand) -> GateState:
        return GateState(gate_id=command.gate_id, state="closed")

    result = await close_security_gate(GateCommand(gate_id="north", reason="drill"))
    assert result.state == "closed"


asyncio.run(main())

Errors from a tool

The SDK returns an uncaught exception to the model as <module>.<qualname>: <message>, truncated to the wire limit. That text reaches the model provider, so never put secrets, credentials, or connection strings in exception messages. To control what the model sees, pass error_transform= on @tool; if it raises or returns a non-string, the SDK fails closed to a generic message rather than leaking the raw exception.

See the API reference for the full @tool signature, and Errors and reliability for how serving and tool failures surface.