Module

server.handler

ASGI handler — translates ASGI scope/messages to chirp types.

The only component that touches raw ASGI directly. Converts scope dicts to typed Request objects, dispatches through middleware and routing, and sends Response back through ASGI send().

6Functions

Functions

compile_middleware_chain 2 Callable[[Request], Any] ▼
Build middleware chain once. Returns async handler(req) -> Response.
def compile_middleware_chain(middleware: tuple[Callable[..., Any], ...], dispatch: Callable[[Request], Any]) -> Callable[[Request], Any]
Parameters
Name Type Description
middleware tuple[Callable[..., Any], ...]
dispatch Callable[[Request], Any]
Returns
Callable[[Request], Any]
create_request_handler 10 Callable[[Request], Any] ▼
Build the full middleware + dispatch chain once. Reuse per request.
def create_request_handler(*, router: Router, middleware: tuple[Callable[..., Any], ...], tool_registry: ToolRegistry | None, mcp_path: str, debug: bool, providers: dict[type, Callable[..., Any]] | None, kida_env: Environment | None, oob_registry: OOBRegistry | None = None, fragment_target_registry: FragmentTargetRegistry | None = None, discovered_routes: list[Any] | None = None) -> Callable[[Request], Any]
Parameters
Name Type Description
router Router
middleware tuple[Callable[..., Any], ...]
tool_registry ToolRegistry | None
mcp_path str
debug bool
providers dict[type, Callable[..., Any]] | None
kida_env Environment | None
oob_registry OOBRegistry | None Default:None
fragment_target_registry FragmentTargetRegistry | None Default:None
discovered_routes list[Any] | None Default:None
Returns
Callable[[Request], Any]
handle_request 17 None ▼
Process a single HTTP request through the full pipeline.
async
async def handle_request(scope: Scope, receive: Receive, send: Send, *, router: Router, middleware: tuple[Callable[..., Any], ...], error_handlers: dict[int | type, Callable[..., Any]], kida_env: Environment | None = None, debug: bool, providers: dict[type, Callable[..., Any]] | None = None, tool_registry: ToolRegistry | None = None, mcp_path: str = '/mcp', sse_heartbeat_interval: float = 15.0, sse_retry_ms: int | None = None, sse_close_event: str | None = None, compiled_handler: Callable[[Request], Any] | None = None, oob_registry: OOBRegistry | None = None, fragment_target_registry: FragmentTargetRegistry | None = None) -> None
Parameters
Name Type Description
scope Scope
receive Receive
send Send
router Router
middleware tuple[Callable[..., Any], ...]
error_handlers dict[int | type, Callable[..., Any]]
kida_env Environment | None Default:None
debug bool
providers dict[type, Callable[..., Any]] | None Default:None
tool_registry ToolRegistry | None Default:None
mcp_path str Default:'/mcp'
sse_heartbeat_interval float Default:15.0
sse_retry_ms int | None Default:None
sse_close_event str | None Default:None
compiled_handler Callable[[Request], Any] | None Default:None
oob_registry OOBRegistry | None Default:None
fragment_target_registry FragmentTargetRegistry | None Default:None
_invoke_handler 8 AnyResponse ▼
Call the matched route handler, converting path params and return value.
async
async def _invoke_handler(match: RouteMatch, request: Request, *, kida_env: Environment | None = None, providers: dict[type, Callable[..., Any]] | None = None, validate_blocks: bool = False, force_inline_sync: bool = False, oob_registry: OOBRegistry | None = None, fragment_target_registry: FragmentTargetRegistry | None = None) -> AnyResponse
Parameters
Name Type Description
match RouteMatch
request Request
kida_env Environment | None Default:None
providers dict[type, Callable[..., Any]] | None Default:None
validate_blocks bool Default:False
force_inline_sync bool Default:False
oob_registry OOBRegistry | None Default:None
fragment_target_registry FragmentTargetRegistry | None Default:None
Returns
AnyResponse
_read_body_if_needed_from_plan 2 dict[str, Any] | None ▼
Pre-read form/JSON body if the handler has extractable dataclass params. Uses …
async
async def _read_body_if_needed_from_plan(plan: InvokePlan | None, request: Request) -> dict[str, Any] | None

Pre-read form/JSON body if the handler has extractable dataclass params.

Uses compiled plan when available.

Parameters
Name Type Description
plan InvokePlan | None
request Request
Returns
dict[str, Any] | None
_read_body_if_needed_inspect 2 dict[str, Any] | None ▼
Fallback: inspect handler for extractable params, read body if needed.
async
async def _read_body_if_needed_inspect(handler: Callable[..., Any], request: Request) -> dict[str, Any] | None
Parameters
Name Type Description
handler Callable[..., Any]
request Request
Returns
dict[str, Any] | None