Skip to main content

​
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: 
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: 
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: 
# When running locally
sb0 run --prompt "test" --extras '{"debug": true}'
Or when calling the API: 
{
  "parts": [{"type": "text", "text": "Hello"}],
  "kwargs": {"debug": true}
}
Print statements are captured as DEBUG logs. Use logger.debug() for cleaner structured logging.

​
Customizing Your Handler

Extend the handler to add custom functionality:

​
Example: Custom Preprocessing

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

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