server.handler

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot
module

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…

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

server.handler

Name Type Default Description
type —
qualified_name —
element_type —
description —
source_file —
line_number —
is_autodoc —
autodoc_element —
_autodoc_template —
_autodoc_url_path —
_autodoc_page_type —
title —
doc_content_hash —

Symbols on this page

function _select_sse_dialect

Select one immutable SSE wire dialect for the connection.

Jump to symbol function compile_middleware_chain

Build middleware chain once. Returns async handler(req) -> Response.

Jump to symbol function _request_path_with_query

Return the request path including the raw query string when present.

Jump to symbol function _cross_shell_boost_redirect

Redirect boosted GETs that would render into the wrong shell.

EmitsHX-Redirectto the destination URL (forcing a full-page load on the client) when…

Jump to symbol function create_request_handler

Build the full middleware + dispatch chain once. Reuse per request.

health_path / ready_path and probe_state (the MutableAppState carrying theready flag + health_checks…

Jump to symbol async function handle_request

Process a single HTTP request through the full pipeline.

Jump to symbol async function _invoke_handler

Call the matched route handler, converting path params and return value.

Jump to symbol async function _read_body_if_needed_from_plan

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

Uses compiled plan when available.

Jump to symbol async function _read_body_if_needed_inspect

Fallback: inspect handler for extractable params, read body if needed.

Jump to symbol
_select_sse_dialect
function 
def _select_sse_dialect(request: Request, manifest: HtmxProvisioningManifest | None) -> Literal['legacy', 'htmx4']

Select one immutable SSE wire dialect for the connection.

Parameters

Name Type Default Description
request Request —
manifest HtmxProvisioningManifest | None —
compile_middleware_chain
function 
def compile_middleware_chain(middleware: tuple[Callable[..., Any], ...], dispatch: Callable[[Request], Any]) -> Callable[[Request], Any]

Build middleware chain once. Returns async handler(req) -> Response.

Parameters

Name Type Default Description
middleware tuple[Callable[..., Any], ...] —
dispatch Callable[[Request], Any] —
_request_path_with_query
function 
def _request_path_with_query(request: Request) -> str

Return the request path including the raw query string when present.

Parameters

Name Type Default Description
request Request —
_cross_shell_boost_redirect
function 
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 Default 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 —
create_request_handler
function 
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, shell_actions_renderer: ShellActionsRenderer | 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', health_path: str | None = None, ready_path: str | None = None, probe_state: Any | None = None, hypermedia_program: HypermediaProgram | None = None) -> Callable[[Request], Any]

Build the full middleware + dispatch chain once. Reuse per request.

health_path / ready_path and probe_state (the MutableAppState carrying theready flag + health_checkslist) wire the auto-mounted ops probes. The probe short-circuit runs BEFORE the middleware chain so /health and /readybypass the secure stack (Session/CSRF/Security headers) and commit teardown entirely — they run no user handler, write no Set-Cookie, and never touch the session.

Parameters

Name Type Default 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 None
fragment_target_registry FragmentTargetRegistry | None None
shell_actions_renderer ShellActionsRenderer | 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'
health_path str | None None
ready_path str | None None
probe_state Any | None None
hypermedia_program HypermediaProgram | None None
handle_request
function 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, max_request_body_size: int | None = None, max_upload_size: int | None = None, upload_spool_threshold: int | None = None, max_upload_parts: int | None = None, compiled_handler: Callable[[Request], Any] | None = None, oob_registry: OOBRegistry | None = None, fragment_target_registry: FragmentTargetRegistry | None = None, shell_actions_renderer: ShellActionsRenderer | None = None, url_for: Callable[..., str] | None = None, debug_wiring: RuntimeDebugWiring | None = None, htmx_manifest: HtmxProvisioningManifest | None = None) -> None

Process a single HTTP request through the full pipeline.

Parameters

Name Type Default Description
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
max_request_body_size int | None None
max_upload_size int | None None
upload_spool_threshold int | None None
max_upload_parts int | None None
compiled_handler Callable[[Request], Any] | None None
oob_registry OOBRegistry | None None
fragment_target_registry FragmentTargetRegistry | None None
shell_actions_renderer ShellActionsRenderer | None None
url_for Callable[..., str] | None None
debug_wiring RuntimeDebugWiring | None None
htmx_manifest HtmxProvisioningManifest | None None
_invoke_handler
function 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, shell_actions_renderer: ShellActionsRenderer | 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, hypermedia_program: HypermediaProgram | None = None) -> AnyResponse

Call the matched route handler, converting path params and return value.

Parameters

Name Type Default Description
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
shell_actions_renderer ShellActionsRenderer | 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
hypermedia_program HypermediaProgram | None None
_read_body_if_needed_from_plan
function 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 Default Description
plan InvokePlan | None —
request Request —
_read_body_if_needed_inspect
function async 
async def _read_body_if_needed_inspect(handler: Callable[..., Any], request: Request) -> dict[str, Any] | None

Fallback: inspect handler for extractable params, read body if needed.

Parameters

Name Type Default Description
handler Callable[..., Any] —
request Request —

View source · /home/runner/work/chirp/chirp/site/../src/chirp/server/handler.py:1