---
eyebrow: Core Concepts
title: Building Agents
description: Learn how to build agents using the Dispatch SDK. Agents are Python functions that respond to events, with automatic schema validation and distributed tracing.
---

## Installation

Install the SDK using uv or pip:

```bash
# Install the Dispatch SDK
uv add git+ssh://git@github.com/datadog-labs/dispatch_agents_sdk.git
```

## The @on Decorator

The `@on` decorator registers a function as a handler for a specific topic. When an event is published to that topic, your handler is invoked with the validated payload.

```python
import dispatch_agents
from dispatch_agents import BasePayload

class GreetingRequest(BasePayload):
    name: str
    language: str = "en"

class GreetingResponse(BasePayload):
    message: str

@dispatch_agents.on(topic="greeting.request")
async def greet(payload: GreetingRequest) -> GreetingResponse:
    if payload.language == "es":
        return GreetingResponse(message=f"Hola, {payload.name}!")
    return GreetingResponse(message=f"Hello, {payload.name}!")
```

> **Handler requirements**: Must be async. First parameter must be a BasePayload subclass. Return type should be a BasePayload subclass (or None).

## Defining Payloads

Payloads define the shape of data your handler accepts and returns. They use Pydantic for validation, giving you type safety and automatic error messages.

```python
from dispatch_agents import BasePayload
from typing import Optional
from datetime import datetime

class TaskPayload(BasePayload):
    task_id: str
    priority: int = 1
    due_date: Optional[datetime] = None
    tags: list[str] = []

class TaskResult(BasePayload):
    task_id: str
    status: str
    completed_at: datetime
```

### Automatic Validation

When a message arrives, Dispatch automatically validates the payload against your input schema. Invalid payloads are rejected before your handler runs:

- Missing required fields return a validation error
- Wrong types are caught and reported
- Extra fields are forbidden by default (strict mode)

## Emitting Events

Agents can emit events to communicate with other agents. Use `emit_event` for fire-and-forget

### Fire and Forget

```python
from dispatch_agents import emit_event

@dispatch_agents.on(topic="order.created")
async def handle_order(payload: OrderPayload) -> None:
    await emit_event(
        topic="notification.send",
        payload={"user_id": payload.user_id, "message": "Order received!"}
    )
```

> **Automatic Context**: When you emit events from within a handler, `trace_id` and `parent_id` are automatically inherited from the current message.

### Direct Function Calls

Use `invoke()` to call a function on another agent directly:

```python
from dispatch_agents import invoke

@dispatch_agents.on(topic="report.request")
async def handle_report_request(payload: ReportRequestPayload) -> ReportResult:
    result = await invoke(
        agent_name="report-generator",
        function_name="generate_report",
        payload={"report_type": payload.type, "date_range": payload.date_range},
        response_model=ReportResponse,
        timeout=120.0
    )
    return ReportResult(status="complete", report_url=result.url)
```

See the [SDK Reference](/docs/sdk) for full `invoke()` documentation.

## Topic Naming Conventions

Dot-separated strings: `resource.action`

- `user.created`
- `order.process`
- `payment.charge`
- `notification.send`

Responses auto-published to `topic.result` or `topic.error`.

## Error Handling & Retries

**Not Retried**: `ValueError` and subclasses -- invalid input, business logic violations, validation failures.

**Auto-Retried**: `OSError` and subclasses -- network timeouts, connection failures, temporary unavailability.

```python
import httpx

@dispatch_agents.on(topic="data.fetch")
async def fetch_data(payload: FetchPayload) -> FetchResult:
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(payload.url)
            return FetchResult(data=response.json())
    except httpx.TimeoutException:
        raise  # OSError subclass - retried
    except httpx.HTTPStatusError as e:
        if e.response.status_code >= 500:
            raise OSError(f"Server error: {e.response.status_code}")
        else:
            raise ValueError(f"Client error: {e.response.status_code}")
```

> **Retry Behavior**: Retries use exponential backoff with jitter. After max retries, a final error message is published to `topic.error`.

## Next Steps

- [CLI Reference](/docs/cli)
- [Architecture](/docs/architecture)