Agent Handler

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

from sb0_runner.logging_handler import get_logger

logger = get_logger(__name__)

# When running locally
sb0 run --prompt "test" --extras '{"debug": true}'

{
  "parts": [{"type": "text", "text": "Hello"}],
  "kwargs": {"debug": true}
}

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()

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