core.base

Defines the fundamental contracts every component in the system must satisfy: typed request/response value objects, the handler abstract base class, a callable-to-handler adapter, and the abstract router interface.

HandlerRequest

@dataclass
class HandlerRequest:
    query:      str
    context:    Dict[str, Any]  = field(default_factory=dict)
    request_id: str             = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp:  float           = field(default_factory=time.time)
    metadata:   Dict[str, Any]  = field(default_factory=dict)

Unified input object passed to every handler.

Fields

Methods

with_metadata(**kwargs) -> HandlerRequest

Returns a new HandlerRequest with kwargs merged into metadata (existing keys are overwritten by kwargs). All other fields (query, context, request_id, timestamp) are copied unchanged — this is an immutable-style update, the original instance is not mutated.

Example:

req = HandlerRequest(query="hello")
req2 = req.with_metadata(route="greet", score=0.92)
assert req.metadata == {}
assert req2.metadata == {"route": "greet", "score": 0.92}

Thread safety: Safe — no shared mutable state is modified; a new object is returned.

Related APIs: HandlerResponse, BaseHandler

HandlerResponse

@dataclass
class HandlerResponse:
    content:          Any
    success:          bool              = True
    error:            Optional[str]     = None
    metadata:         Dict[str, Any]    = field(default_factory=dict)
    processing_time:  float             = 0.0

Unified output object returned by every handler.

Fields

Class methods

HandlerResponse.ok(content, **meta) -> HandlerResponse

Convenience constructor for a successful response: success=True, error=None, and metadata=meta (all keyword arguments become metadata entries).

return HandlerResponse.ok("The answer is 42", source="calculator", tokens=12)

HandlerResponse.fail(error: str, **meta) -> HandlerResponse

Convenience constructor for a failed response: content=None, success=False, error=error, metadata=meta.

return HandlerResponse.fail("Upstream API timed out", retry_after=5)

Magic methods

__bool__(self) -> bool

Returns self.success. Enables if response: as shorthand for if response.success:.

Related APIs: HandlerRequest, RoutingResult

BaseHandler

class BaseHandler(ABC):
    @abstractmethod
    def handle(self, request: HandlerRequest) -> HandlerResponse: ...

Abstract base for all route handlers. Every handler must implement handle(); the async variant defaults to calling the sync version.

Overview

BaseHandler exists so handlers are explicit, testable, mockable objects rather than bare callables, with request/response contracts that are typed and IDE-friendly, and with async/streaming as first-class, optional capabilities layered on top of a mandatory sync path.

Abstract methods (must implement)

handle(self, request: HandlerRequest) -> HandlerResponse

Synchronously process a request. This is the only method a subclass is required to implement.

Raises: Whatever the implementation raises — Route.execute catches all exceptions from this call and converts them into HandlerResponse.fail(f"Handler raised {type(exc).__name__}: {exc}"), so handlers do not need their own top-level try/except purely for router safety (though doing so lets you control the error message).

Optional overrides

handle_async(self, request: HandlerRequest) -> HandlerResponse (async)

Default implementation: return self.handle(request) — i.e. it runs the sync path with no real concurrency benefit. Override this for true async I/O (LLM calls, DB queries, network requests).

validate(self, request: HandlerRequest) -> bool

Pre-execution validation hook. Default: always returns True. Return False to abort execution — Route.execute / execute_async check this before calling handle/handle_async, and if it returns False, execution short-circuits to HandlerResponse.fail(f"Handler validation failed for route '{self.name}'.") without ever calling handle. In SEQUENTIAL execution mode, a validation failure lets the engine fall through to the next candidate.

stream(self, request: HandlerRequest) -> Iterator[str]

Yield response chunks for streaming. Default implementation calls handle() and yields the entire str(content) as a single chunk — override for genuine token-level streaming.

stream_async(self, request: HandlerRequest) -> AsyncIterator[str] (async generator)

Async streaming variant; default calls handle_async() and yields the full content as one chunk.

Other methods

__call__(self, request: HandlerRequest) -> HandlerResponse

Allows a handler instance to be invoked directly, e.g. my_handler(request) is equivalent to my_handler.handle(request).

__repr__(self) -> str

Returns f"{self.__class__.__name__}()".

Complete example: a custom handler with async I/O and real streaming

import httpx
from zerda.router import BaseHandler, HandlerRequest, HandlerResponse

class WeatherAPIHandler(BaseHandler):
    def __init__(self, api_key: str):
        self._api_key = api_key

    def validate(self, request: HandlerRequest) -> bool:
        return "city" in request.context

    def handle(self, request: HandlerRequest) -> HandlerResponse:
        # Synchronous fallback path
        resp = httpx.get(
            "https://api.example.com/weather",
            params={"city": request.context["city"], "key": self._api_key},
        )
        if resp.status_code != 200:
            return HandlerResponse.fail(f"Weather API returned {resp.status_code}")
        return HandlerResponse.ok(resp.json())

    async def handle_async(self, request: HandlerRequest) -> HandlerResponse:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                "https://api.example.com/weather",
                params={"city": request.context["city"], "key": self._api_key},
            )
        if resp.status_code != 200:
            return HandlerResponse.fail(f"Weather API returned {resp.status_code}")
        return HandlerResponse.ok(resp.json())

    async def stream_async(self, request: HandlerRequest):
        response = await self.handle_async(request)
        for word in str(response.content).split():
            yield word + " "

Best practices

• Override validate for cheap, request-shape checks (missing required context keys) rather than raising inside handle — this lets SEQUENTIAL execution mode gracefully try the next candidate.
• Prefer returning HandlerResponse.fail(...) over raising exceptions when the failure is expected/business-logic (e.g. "no results found") — reserve exceptions for genuinely unexpected errors, since the router will catch and stringify them either way.
• Override handle_async whenever your handler does I/O; the default delegation to handle() provides no concurrency benefit under PARALLEL execution mode.

Common mistakes

• Forgetting to override handle_async and then using ExecutionMode.PARALLEL expecting true concurrency — with the default implementation, "parallel" execution of I/O-bound handlers still blocks per-handler since each handle_async call just runs the synchronous handle().
• Mutating request.context in place inside a handler when other code (e.g. ExecutionEngine.chain) may pass the same context dict to subsequent steps — prefer treating context as read-only within a single handler invocation.

Related APIs: CallableHandler, Route.execute, ExecutionEngine

CallableHandler

class CallableHandler(BaseHandler):
    def __init__(self, fn): ...

Wraps a plain Callable(query: str) -> Any so that simple function-based handlers work without subclassing BaseHandler. This is a backward-compatibility shim.

Constructor

Parameters:

Raises: TypeError — if fn is not callable (f"Expected a callable, got {type(fn).__name__}").

Methods

handle(self, request: HandlerRequest) -> HandlerResponse

Calls self._fn(request.query) inside a try/except. On success, wraps the return value in HandlerResponse.ok(result). On any exception, returns HandlerResponse.fail(str(exc)).

__repr__(self) -> str

Returns f"CallableHandler({name})" where name is fn.__name__ if available, else repr(fn).

Examples

from zerda.router import CallableHandler, HandlerRequest

# Basic
handler = CallableHandler(lambda q: {"answer": q})
response = handler.handle(HandlerRequest(query="hi"))
assert response.content == {"answer": "hi"}

# Any Route(handler=...) or RouteGroup.route(...) call that receives a plain
# callable wraps it in CallableHandler automatically — you rarely construct
# this class directly.

Related APIs: BaseHandler, Route

BaseRouter

class BaseRouter(ABC):
    @abstractmethod
    def route(self, request: HandlerRequest) -> "RoutingResult": ...
    @abstractmethod
    async def route_async(self, request: HandlerRequest) -> "RoutingResult": ...
    @abstractmethod
    def register(self, route: "Route") -> None: ...

Abstract contract that both the top-level HierarchicalRouter and any future nested sub-routers must satisfy, allowing sub-routers to be nested arbitrarily deep. Not directly instantiable — HierarchicalRouter is the sole implementation found in the analyzed source.

Related APIs: HierarchicalRouter