> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sb0.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Agent Handler

> Learn how to implement your agent's core logic

## Handler Overview

Your agent's logic is defined in a `Handler` class in `handle_query.py`. The handler receives user messages and yields responses using the Claude Agent SDK.

## Handler Structure

The handler implements a single async method:

```python theme={null}
from typing import AsyncIterator, Any
from claude_agent_sdk import ClaudeAgentOptions, query, Message
from sb0_runner.logging_handler import get_logger

logger = get_logger(__name__)

class Handler:
    def __init__(self) -> None:
        """Initialize handler.

        Called once when the handler is loaded.
        Use this to initialize shared resources like API clients or DB connections.
        """
        logger.info("Handler initialized")

    async def handle_query(
        self,
        prompt: str,
        session_id: str | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Message]:
        """Stream Claude responses.

        Args:
            prompt: The user's input prompt.
            session_id: Optional session ID for conversation continuity.
            **kwargs: Additional parameters from the request, which may include:
                - ANTHROPIC_API_KEY (str): API key from environment
                - debug (bool): Enable debug log output
                - Any custom fields from your middleware

        Yields:
            Message: Claude SDK message objects.
        """
        logger.info("Processing query, session_id=%s", session_id)

        opts = ClaudeAgentOptions(
            include_partial_messages=True,
            model="claude-haiku-4-5"
        )

        if session_id:
            opts.resume = session_id
            logger.debug("Resuming session: %s", session_id)

        async for message in query(prompt=prompt, options=opts):
            logger.debug("Yielding message type=%s", type(message).__name__)
            yield message

        logger.info("Query completed")
```

## Key Components

### The `handle_query` Method

* **`prompt`** (str) - The user's input message
* **`session_id`** (str | None) - Optional session ID for conversation continuity
* **`kwargs`** (Any) - Additional parameters from the request
* **Returns** - AsyncIterator\[Message] that yields Claude SDK messages

### Claude Agent Options

Configure the Claude Agent SDK behavior:

* **`include_partial_messages`** (bool) - Stream partial responses as they're generated
* **`model`** (str) - Claude model to use (e.g., "claude-haiku-4-5", "claude-sonnet-4")
* **`resume`** (str) - Session ID to resume a previous conversation

## Structured Logging

The handler includes a structured logging system:

```python theme={null}
from sb0_runner.logging_handler import get_logger

logger = get_logger(__name__)
```

### Log Levels

* **`logger.debug()`** - Development debugging, verbose information
* **`logger.info()`** - Important operational events
* **`logger.warning()`** - Warning messages
* **`logger.error()`** - Error messages

### Viewing Logs

By default, only yielded messages reach clients. To see logs during development, enable debug mode:

```bash theme={null}
# When running locally
sb0 run --prompt "test" --extras '{"debug": true}'
```

Or when calling the API:

```json theme={null}
{
  "parts": [{"type": "text", "text": "Hello"}],
  "kwargs": {"debug": true}
}
```

<Note>
  Print statements are captured as DEBUG logs. Use `logger.debug()` for cleaner structured logging.
</Note>

## Customizing Your Handler

Extend the handler to add custom functionality:

### Example: Custom Preprocessing

```python theme={null}
class Handler:
    async def handle_query(
        self,
        prompt: str,
        session_id: str | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Message]:
        # Preprocess the prompt
        processed_prompt = self._preprocess(prompt)

        opts = ClaudeAgentOptions(include_partial_messages=True)
        if session_id:
            opts.resume = session_id

        async for message in query(prompt=processed_prompt, options=opts):
            yield message

    def _preprocess(self, prompt: str) -> str:
        """Custom preprocessing logic"""
        return prompt.strip().lower()
```

### Example: External API Integration

```python theme={null}
import httpx

class Handler:
    def __init__(self) -> None:
        self.client = httpx.AsyncClient()

    async def handle_query(
        self,
        prompt: str,
        session_id: str | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Message]:
        # Fetch external data
        response = await self.client.get("https://api.example.com/data")
        context = response.json()

        # Include context in prompt
        enhanced_prompt = f"{prompt}\n\nContext: {context}"

        opts = ClaudeAgentOptions(include_partial_messages=True)
        if session_id:
            opts.resume = session_id

        async for message in query(prompt=enhanced_prompt, options=opts):
            yield message
```

## Next Steps

* Learn about [agent configuration](/v0-docs/agents/configuration)
* Understand how to [add dependencies](/v0-docs/agents/dependencies)