Module

app

Chirp application class.

Mutable during setup (route registration, middleware, filters, tools). Frozen at runtime when app.run() or call() is first invoked.

Classes

_PendingRoute 4
A route waiting to be compiled.

A route waiting to be compiled.

Attributes

Name Type Description
path str
handler Handler
methods list[str] | None
name str | None
_PendingTool 3
A tool waiting to be compiled.

A tool waiting to be compiled.

Attributes

Name Type Description
name str
description str
handler Callable[..., Any]
App 25
The chirp application. Mutable during setup (route registration, middleware, filters). Frozen at r…

The chirp application.

Mutable during setup (route registration, middleware, filters). Frozen at runtime whenapp.run() or __call__()is first invoked.

Thread safety:

The setup phase is single-threaded (decorators at import time).
The freeze transition uses a Lock + double-check to ensure exactly
one thread compiles the app, even under free-threading where
multiple ASGI workers could call ``__call__()`` concurrently on
first request.

Methods

db 0 Database
The database instance, if configured. Raises ``RuntimeError`` if no database w…
property
def db(self) -> Database

The database instance, if configured.

RaisesRuntimeErrorif no database was configured on this app.

Usage::

app = App(db="sqlite:///app.db")

@app.route("/users")
async def users():
    return await app.db.fetch(User, "SELECT * FROM users")
Returns
Database
tool_events 0 ToolEventBus
Public access to the tool event bus for SSE subscriptions. Usage:: @app.r…
property
def tool_events(self) -> ToolEventBus

Public access to the tool event bus for SSE subscriptions.

Usage::

@app.route("/dashboard/feed")
async def feed(request: Request):
    async def stream():
        async for event in app.tool_events.subscribe():
            yield Fragment("dashboard.html", "row", event=event)
    return EventStream(stream())
Returns
ToolEventBus
route 3 Callable[[Handler], Hand…
Register a route handler via decorator.
def route(self, path: str, *, methods: list[str] | None = None, name: str | None = None) -> Callable[[Handler], Handler]
Parameters
Name Type Description
path
methods Default:None
name Default:None
Returns
Callable[[Handler], Handler]
provide 2
Register a provider factory for dependency injection. When a handler parameter…
def provide(self, annotation: type, factory: Callable[..., Any]) -> None

Register a provider factory for dependency injection.

When a handler parameter's type annotation matches annotation, chirp calls factory (with no arguments) and injects the result.

Works for both@app.routeand filesystem page handlers::

app.provide(DocumentStore, get_store)

# Any handler with ``store: DocumentStore`` gets it injected:
def get(store: DocumentStore) -> Page: ...
Parameters
Name Type Description
annotation

The type annotation to match against handler params.

factory

A zero-argument callable that returns the service instance.

mount_pages 1
Mount a filesystem-based pages directory. Walks the directory and registers ro…
def mount_pages(self, pages_dir: str | None = None) -> None

Mount a filesystem-based pages directory.

Walks the directory and registers routes for everypage.py and handler.pyfile, with automatic layout nesting and context cascade.

_layout.html files define shells with {% block content %} and{# target: element_id #}declarations. _context.pyfiles provide shared context that cascades from parent directories to children.

Parameters
Name Type Description
pages_dir

Path to the pages directory. Defaults to"pages"relative to the working directory.

Default:None
tool 2 Callable[[Callable[..., …
Register a function as an MCP tool via decorator. The tool is callable by MCP …
def tool(self, name: str, *, description: str = '') -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a function as an MCP tool via decorator.

The tool is callable by MCP clients via JSON-RPC at/mcp and can also be called directly from route handlers as a normal Python function.

Tool calls emitToolCallEvent to app.tool_events for real-time dashboard subscriptions.

Usage::

@app.tool("search", description="Search inventory")
async def search(query: str, category: str | None = None) -> list[dict]:
    return await db.search(query, category=category)
Parameters
Name Type Description
name
description Default:''
Returns
Callable[[Callable[..., Any]], Callable[..., Any]]
error 1 Callable[[ErrorHandler],…
Register an error handler via decorator.
def error(self, code_or_exception: int | type[Exception]) -> Callable[[ErrorHandler], ErrorHandler]
Parameters
Name Type Description
code_or_exception
Returns
Callable[[ErrorHandler], ErrorHandler]
add_middleware 1
Add a middleware to the pipeline.
def add_middleware(self, middleware: Middleware) -> None
Parameters
Name Type Description
middleware
template_filter 1 Callable[[Callable[..., …
Register a kida template filter.
def template_filter(self, name: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]
Parameters
Name Type Description
name Default:None
Returns
Callable[[Callable[..., Any]], Callable[..., Any]]
template_global 1 Callable[[Callable[..., …
Register a kida template global.
def template_global(self, name: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]
Parameters
Name Type Description
name Default:None
Returns
Callable[[Callable[..., Any]], Callable[..., Any]]
on_startup 1 Callable[..., Any]
Register an async or sync startup hook via decorator. Hooks run in registratio…
def on_startup(self, func: Callable[..., Any]) -> Callable[..., Any]

Register an async or sync startup hook via decorator.

Hooks run in registration order during ASGI lifespan startup, before the server begins accepting HTTP requests.

Usage::

@app.on_startup
async def setup():
    await db.connect()
Parameters
Name Type Description
func
Returns
Callable[..., Any]
on_shutdown 1 Callable[..., Any]
Register an async or sync shutdown hook via decorator. Hooks run in registrati…
def on_shutdown(self, func: Callable[..., Any]) -> Callable[..., Any]

Register an async or sync shutdown hook via decorator.

Hooks run in registration order during ASGI lifespan shutdown, after the server stops accepting new requests.

Usage::

@app.on_shutdown
async def teardown():
    await db.disconnect()
Parameters
Name Type Description
func
Returns
Callable[..., Any]
on_worker_startup 1 Callable[..., Any]
Register a per-worker startup hook via decorator. Unlike ``on_startup`` (which…
def on_worker_startup(self, func: Callable[..., Any]) -> Callable[..., Any]

Register a per-worker startup hook via decorator.

Unlikeon_startup(which runs once during ASGI lifespan), this hook runs on each worker's event loop before that worker begins accepting connections. Use this for async resources that bind to the event loop (httpx clients, DB connection pools, etc.).

Requires a server that sendspounce.worker.startupscopes (e.g. pounce). Under other servers, the hooks simply never fire.

Usage::

@app.on_worker_startup
async def create_client():
    _client_var.set(httpx.AsyncClient(...))
Parameters
Name Type Description
func
Returns
Callable[..., Any]
on_worker_shutdown 1 Callable[..., Any]
Register a per-worker shutdown hook via decorator. Runs **on each worker's eve…
def on_worker_shutdown(self, func: Callable[..., Any]) -> Callable[..., Any]

Register a per-worker shutdown hook via decorator.

Runs on each worker's event loop after that worker stops accepting connections. Pair withon_worker_startupfor per-worker resource cleanup.

Usage::

@app.on_worker_shutdown
async def close_client():
    client = _client_var.get(None)
    if client:
        await client.aclose()
Parameters
Name Type Description
func
Returns
Callable[..., Any]
run 3
Start the development server via pounce. Compiles the app (freezing routes, mi…
def run(self, host: str | None = None, port: int | None = None, *, lifecycle_collector: object | None = None) -> None

Start the development server via pounce.

Compiles the app (freezing routes, middleware, templates) and starts serving requests with auto-reload enabled.

Parameters
Name Type Description
host

Override bind host.

Default:None
port

Override bind port.

Default:None
lifecycle_collector

Optional Pounce LifecycleCollector for observability. Forwarded to the Pounce Server.

Default:None
check 0
Validate the hypermedia surface and print results. Freezes the app if needed, …
def check(self) -> None

Validate the hypermedia surface and print results.

Freezes the app if needed, then checks that every htmx target, form action, and fragment contract resolves to a valid route. RaisesSystemExit(1)if errors are found.

Usage::

app.check()  # prints results and exits on errors
Internal Methods 9
__init__ 3
def __init__(self, config: AppConfig | None = None, *, db: Database | str | None = None, migrations: str | None = None) -> None
Parameters
Name Type Description
config Default:None
db Default:None
migrations Default:None
_register_page_handler 5
Register a single page handler with cascade context wrapper.
def _register_page_handler(self, *, url_path: str, handler: Callable[..., Any], methods: list[str], layout_chain: Any, context_providers: tuple[Any, ...]) -> None
Parameters
Name Type Description
url_path
handler
methods
layout_chain
context_providers
__call__ 3
ASGI 3.0 entry point. Handles lifespan and per-worker lifecycle scopes directl…
async
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None

ASGI 3.0 entry point.

Handles lifespan and per-worker lifecycle scopes directly, then delegates HTTP scopes to the request handler pipeline.

Parameters
Name Type Description
scope
receive
send
_handle_lifespan 3
Run the ASGI lifespan protocol. Freezes the app at startup (before first HTTP …
async
async def _handle_lifespan(self, scope: Scope, receive: Receive, send: Send) -> None

Run the ASGI lifespan protocol.

Freezes the app at startup (before first HTTP request), then runs registered startup/shutdown hooks and signals completion back to the server.

Parameters
Name Type Description
scope
receive
send
_handle_worker_startup 0
Run registered per-worker startup hooks. Called by pounce when a worker thread…
async
async def _handle_worker_startup(self) -> None

Run registered per-worker startup hooks.

Called by pounce when a worker thread starts, on that worker's event loop. Errors propagate to pounce, which prevents the worker from accepting connections.

_handle_worker_shutdown 0
Run registered per-worker shutdown hooks. Called by pounce when a worker threa…
async
async def _handle_worker_shutdown(self) -> None

Run registered per-worker shutdown hooks.

Called by pounce when a worker thread stops, on that worker's event loop. Errors propagate to pounce, which logs and continues shutdown.

_ensure_frozen 0
Thread-safe freeze with double-check locking. Under free-threading (3.14t), mu…
def _ensure_frozen(self) -> None

Thread-safe freeze with double-check locking.

Under free-threading (3.14t), multiple ASGI worker threads could call call() concurrently on first request. This pattern ensures exactly one thread performs compilation.

_freeze 0
Compile the app into its frozen runtime state. MUST only be called while holdi…
def _freeze(self) -> None

Compile the app into its frozen runtime state.

MUST only be called while holding _freeze_lock.

_run_debug_checks 0
Run contract checks and print results to stderr. Uses the rich terminal format…
def _run_debug_checks(self) -> None

Run contract checks and print results to stderr.

Uses the rich terminal formatter so output matches the pounce startup banner's visual language. Printed even when there are no issues (confirms checks ran).