SDK Tool Host: declare, serve, and operate Tools

Use the SDK Tool Host to let Tasks call typed Python functions in your application. An SDK Tool is one such function, declared with @tool. sdk.serve() publishes the Tool Host’s manifest, the set of Tool definitions it offers, and maintains heartbeats. Tool Calls arrive through Task event streams opened by that same SDK instance; serve() alone does not receive them.

The Tool Host is separate from a customer-hosted Duale AI Platform deployment.

For an existing remote MCP server, see Tools and connect an MCP server. MCP Connections run outside your SDK process.

Publish and verify a Tool Host

Before you serve a Tool

Complete Install the SDK before you configure a Tool Host.

Check these three conditions before registration:

• Set DUALEAI_TOKEN and the matching, centrally provisioned DUALEAI_AGENT_ID. The SDK never creates Agent Identities; serving Tools without an Agent Identity identifier raises an error.
• For each @tool, provide 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.

The SDK rejects a missing value or unsupported type instead of choosing one for you.

Declare and serve a Tool

This example publishes one Tool and maintains heartbeats. It submits no Task and executes no Tool Call.

import asyncio
from datetime import timedelta

from pydantic import BaseModel, ConfigDict

from dualeai import DualeAIConfig, DualeAISDK, tool


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

    sku: str


class CatalogItem(BaseModel):
    sku: str
    availability: str


async def main() -> None:
    sdk = DualeAISDK(config=DualeAIConfig(), auto_start=False)

    @tool(
        sdk=sdk,
        description="Look up a product in the local catalog.",
        timeout=timedelta(seconds=10),
    )
    async def lookup_product(query: CatalogQuery) -> CatalogItem:
        return CatalogItem(sku=query.sku, availability="available")

    async with sdk:
        await sdk.start()
        print("Tool Host lifecycle started")
        await sdk.serve()


asyncio.run(main())

Verify the Tool Host

Save the example as tool_host.py, then start it with debug logging:

DUALEAI_DEBUG=true python tool_host.py

The script prints Tool Host lifecycle started after registration and the initial heartbeat succeed. Debug logs can show connection and heartbeat activity. Stop the example with Ctrl+C after you verify startup.

To verify execution, use the same SDK instance for registration and ask(). The Tool execution sample (opens in a new tab) shows the full flow with synthetic stock data. Record an invocation inside the Tool; registration, heartbeats, and a completed Task do not prove that the model saw or called it.

Define what the model can call

When a Tool is exposed to a model, the model receives its name, description=, and parameter JSON Schema. A Python docstring is not sent. Use annotations for parameter types, constraints, and hints:

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")]

Your function returns the Tool Result, or Duale AI returns a rejection or a timeout result instead. The Task can use the result on later model turns.

The SDK uses the annotations for both the schema sent to the model and argument validation. It validates calls against the current function definition. After a manifest change, that definition can differ from the one the model saw. Treat that mismatch as a failed Tool Call.

Use Pydantic models for parameters. Other nested classes still validate locally but advertise no constraints to the model.

Reject input you did not declare

The SDK rejects unexpected top-level parameters. Nested models use their own Pydantic settings, which silently drop unknown fields by default. The Tool then runs without reporting those fields as invalid.

Rejecting unknown fields on the outer model does not reject them inside nested models. Make every model inherit from 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

For a model owned by another library, accept your own strict model and construct the vendor object inside the function.

Keep the Tool Result small

Return an identifier, a summary, or the relevant rows. Tool Results become part of what the model receives on later turns.

Keep results far below the SDK Tool Result size limit; a larger submission fails. Put large documents in a Library for Library search.

Run and change the Tool Host

sdk.serve() publishes the current Tool set, maintains heartbeats, and attempts deregistration on shutdown. It blocks until shutdown or a lifecycle failure, so supervise it as a long-running application service. The API reference defines cooperative shutdown.

Duale AI validates the manifest when serve() publishes it. A rejection raises BusinessError without retrying. See Tool limits for manifest and schema limits, and Tool Host errors for the rejection codes.

Operate Tools explains how Duale AI names an SDK Tool for the model and resolves cross-Source collisions.

Change a Tool without breaking active work

Treat a registered Tool name and parameter schema as a versioned interface. Duale AI does not pin a manifest version to a Task or Tool Call. Do not run incompatible manifests concurrently under one Agent Identity. Make compatible changes in place. For a breaking input change, publish a new Tool name. Every process that handles Task streams must support both names. Remove the old name only after every Task that can use its definition has passed its deadline and its Effects are reconciled.

Handle delivery and retries

A Tool Call can fail or time out before reaching your code, or the same (task_id, tool_call_id) can arrive more than once. Duplicate suppression covers only one DualeAISDK instance; another instance or a restarted process can run the function again. After a timeout, whether the function ran or caused an Effect remains uncertain.

For a state-changing Tool, do not use the SDK’s process-local duplicate cache as an Effect ledger. Secure integration defines the procedure for recording attempted actions durably and checking uncertain outcomes, including the identifiers to retain.

Retries are opt-in and default to 0. All attempts share one absolute Tool Call deadline. A retry does not extend the configured timeout and cannot start after that deadline. A synchronous function already running in a worker thread can continue after the SDK stops waiting. A timeout therefore does not tell you whether an Effect occurred. Enable retries only for retry-safe Tools. The API reference defines the decorator arguments.

Inside the function, current_tool_context() returns a read-only context with the Task ID, the Tool Call ID, the attempt number for this delivery, and the deadline for returning the result. See Tool context fields. Read a Tool Call outcome explains how each call’s outcome appears in the Dashboard.

Test and diagnose the function

A @tool-decorated function remains directly callable. Unit-test its domain logic and return value as you would any other Python function. A direct call bypasses the SDK’s argument validation and Tool Result serialization. Test those through Tool dispatch with an injected transport; MockSDK explains how to supply test responses. Publication, delivery, and Tool Result submission also need their own integration checks.

Never put secrets, credentials, or connection strings in exception messages. error_transform= controls the rendered Tool Result text, but it does not redact the original exception from Tool Host logs. Troubleshoot the SDK covers serving and Tool failures.