Classes
_PendingRoute
4
▼
A route waiting to be compiled.
_PendingRoute
4
▼
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.
_PendingTool
3
▼
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…
App
25
▼
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
db
0
Database
▼
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
tool_events
0
ToolEventBus
▼
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.
route
3
Callable[[Handler], Hand…
▼
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…
provide
2
▼
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…
mount_pages
1
▼
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 None
|
tool
2
Callable[[Callable[..., …
▼
Register a function as an MCP tool via decorator.
The tool is callable by MCP …
tool
2
Callable[[Callable[..., …
▼
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.
error
1
Callable[[ErrorHandler],…
▼
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.
add_middleware
1
▼
def add_middleware(self, middleware: Middleware) -> None
Parameters
| Name | Type | Description |
|---|---|---|
middleware |
— |
template_filter
1
Callable[[Callable[..., …
▼
Register a kida template filter.
template_filter
1
Callable[[Callable[..., …
▼
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.
template_global
1
Callable[[Callable[..., …
▼
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…
on_startup
1
Callable[..., Any]
▼
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…
on_shutdown
1
Callable[..., Any]
▼
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…
on_worker_startup
1
Callable[..., Any]
▼
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…
on_worker_shutdown
1
Callable[..., Any]
▼
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…
run
3
▼
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, …
check
0
▼
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
▼
__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.
_register_page_handler
5
▼
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
__call__
3
▼
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
_handle_lifespan
3
▼
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
_handle_worker_startup
0
▼
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
_handle_worker_shutdown
0
▼
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…
_ensure_frozen
0
▼
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…
_freeze
0
▼
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…
_run_debug_checks
0
▼
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).