---
title: "Declare and serve Python tools with the SDK"
description: "The Duale AI SDK lets you declare Python functions with @tool and publish them via sdk.serve() so a task can call them."
lang: en
status: public-preview
lastUpdated: 2026-09-03
url: https://dev.duale.ai/en/docs/sdk/tools
---

## AI-generated summary

Declare Python functions as model-callable tools with the @tool decorator, then publish them via sdk.serve() for a configured agent.

- Requires DUALE_TOKEN, DUALE_AGENT_ID, and Pydantic-serializable inputs before calling serve().
- The serve lifecycle covers registration, heartbeat every 60 seconds, and best-effort deregistration on shutdown.
- Tool delivery is at-least-once; deduplicate by task_id and tool_call_id within a running process.
- One agent serves at most 64 tools with timeouts capped at 3600 seconds.
- Retries are opt-in per tool, default to 0, and respect both timeout and deadline bounds.

Summaries were generated by AI. Generative AI is experimental.

---

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:

```python runnable
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:

```python
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:

```python
# 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](https://dev.duale.ai/en/docs/security/secure-integration.md) owns the durable claim procedure and both ways to handle a repeatable effect.

## Retries

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

```python
# 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](#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:

```python runnable
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](https://dev.duale.ai/en/docs/sdk/reference.md) for the full `@tool` signature, and [Errors and reliability](https://dev.duale.ai/en/docs/sdk/errors.md) for how serving and tool failures surface.

## Related content

- [Python SDK for bounded agent work with typed results](https://dev.duale.ai/en/docs/sdk.md)
- [Application-facing SDK API reference](https://dev.duale.ai/en/docs/sdk/reference.md)
- [SDK task lifecycle, routing, and streaming](https://dev.duale.ai/en/docs/sdk/concepts.md)
- [Manage Libraries and documents with the SDK](https://dev.duale.ai/en/docs/sdk/manage-libraries.md)
- [Build an agent that passes review](https://dev.duale.ai/en/docs/security/secure-integration.md)
- [Production runtime for durable AI agents](https://dev.duale.ai/index.md)

---

## Sitemap

See the full [Markdown sitemap](https://dev.duale.ai/sitemap.md) for all pages.