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

8Functions

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]
_request_path_with_query 1 str ▼
Return the request path including the raw query string when present.
def _request_path_with_query(request: Request) -> str
Parameters
Name Type Description
request Request
Returns
str
_cross_shell_boost_redirect 6 Response | None ▼
Redirect boosted GETs that would render into the wrong shell. Emits ``HX-Redir…
def _cross_shell_boost_redirect(request: Request, match: RouteMatch, *, router: Router, route_layout_chains: Mapping[str, Any] | None, fragment_target_registry: FragmentTargetRegistry | None, swap_scope_map: Mapping[str, str] | None) -> Response | None

Redirect boosted GETs that would render into the wrong shell.

EmitsHX-Redirectto the destination URL (forcing a full-page load on the client) when the framework cannot guarantee a correct fragment swap. The policy is conservative: only boosted GETs are affected, and we always prefer redirect over rendering a fragment into a target the destination shell cannot satisfy.

Cases handled:

1. App has shell configured but registries are inconsistent
   (framework setup bug) — redirect rather than render broken.
2. Current and destination have separate layout chains with no
   shared navigation ancestor (true cross-shell) — redirect.
3. Computed swap target does not match the client's ``HX-Target``
   (existing behavior).

Apps without app-shell (emptyswap_scope_map) are unaffected: their boosted responses pass through to normal fragment rendering.

Parameters
Name Type Description
request Request
match RouteMatch
router Router
route_layout_chains Mapping[str, Any] | None
fragment_target_registry FragmentTargetRegistry | None
swap_scope_map Mapping[str, str] | None
Returns
Response | None
create_request_handler 15 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, route_layout_chains: Mapping[str, Any] | None = None, swap_scope_map: Mapping[str, str] | None = None, discovered_routes: list[Any] | None = None, debug_wiring: RuntimeDebugWiring | None = None, suspense_error_template: str | None = None, suspense_error_block: str = 'fallback') -> 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
route_layout_chains Mapping[str, Any] | None Default:None
swap_scope_map Mapping[str, str] | None Default:None
discovered_routes list[Any] | None Default:None
debug_wiring RuntimeDebugWiring | None Default:None
suspense_error_template str | None Default:None
suspense_error_block str Default:'fallback'
Returns
Callable[[Request], Any]
handle_request 19 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, url_for: Callable[..., str] | None = None, debug_wiring: RuntimeDebugWiring | 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
url_for Callable[..., str] | None Default:None
debug_wiring RuntimeDebugWiring | None Default:None
_invoke_handler 14 AnyResponse ▼
Call the matched route handler, converting path params and return value.
async
async def _invoke_handler(match: RouteMatch, request: Request, *, router: Router, 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, route_layout_chains: Mapping[str, Any] | None = None, swap_scope_map: Mapping[str, str] | None = None, suspense_error_template: str | None = None, suspense_error_block: str = 'fallback', fragment_block: str | None = None) -> AnyResponse
Parameters
Name Type Description
match RouteMatch
request Request
router Router
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
route_layout_chains Mapping[str, Any] | None Default:None
swap_scope_map Mapping[str, str] | None Default:None
suspense_error_template str | None Default:None
suspense_error_block str Default:'fallback'
fragment_block str | 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