Applications

Mountable ASGI application types.

AsgiApplication

Base class for all genro-asgi applications.

AsgiApplication package.

class genro_asgi.applications.asgi_application.AsgiApplication(**kwargs)

Bases: RoutingClass

Base class for apps mounted on AsgiServer.

Routes live on the app’s single router (the lazy route property) and provide a default index() method. Subclasses define openapi_info for metadata and add routes with the @route() decorator; sub-trees are built by attaching child RoutingClass instances.

Example:

class MyApp(AsgiApplication):
    openapi_info = {"title": "My API", "version": "1.0.0"}

    @route()
    def hello(self):
        return "Hello!"

    def on_init(self, **kwargs):
        # Sub-tree under /backoffice via a child instance
        self.attach_instance(Backoffice(self), name="backoffice")

__init__(**kwargs)

Set up mount name, app path, db registry and middleware slot, then call on_init.

Parameters:

**kwargs (Any) – base_dir and db_name are consumed here; the rest is forwarded to on_init().

property app_path: Path

Filesystem directory where this app is located.

Defaults to the directory of the module that defines the app’s class; the server may override it at mount time. Resources, templates and other per-app files are resolved relative to this path.

app_protocol: ClassVar[str] = 'asgi'

authenticate(scope)

Resolve the identity for a request. Default: delegate to the server.

Override to give an app its own authentication. Returning the server’s auth keeps the current shared behaviour.

Return type:

dict[str, Any] | None

bind_kwargs(node, request)

Return the kwargs to call node with. Base: the request kwargs.

Subclasses with a wire dialect (REST/JSON) override to reconcile payload conventions with the handler signature (see OpenApiApplication).

Return type:

dict[str, Any]

configure(node)

Interpret this app’s own configuration node, after mounting.

The app receives its whole application node (attributes already used as constructor kwargs, plus children) and interprets the tags of its own grammar. The server does not interpret an app’s config: an app may extend the grammar with tags the core does not know.

Called by the server right after attach_instance (so self.server is available). A databases child builds handlers via the server’s build_db_handler and fills self.db_registry; the middleware attribute builds this app’s own middleware chain (self._chain).

Return type:

None

property db: Any

Default db handler for this app (resolves db_name or “default”).

default_db_name: ClassVar[str | None] = None

get_db(name)

Return a db handler by name: the app registry first, then the server.

Returns None when no database is registered under name in either registry (the app’s own databases shadow the server’s shared ones).

Return type:

Any

async handle_request(scope, receive, send)

Resolve the request in this app’s router and send the response.

This is the per-app routing engine: authenticate, resolve the handler in self.route from the path relative to the mount, build the context, execute and respond. Subclasses with a foreign protocol (e.g. a WSGI proxy) override this entirely.

Return type:

None

index()

Return HTML splash page. Override for custom index.

Return type:

str

load_resource(*args, name)

Load a resource file as (content_bytes, mime_type) or None.

Mounted: tries the server’s ResourceLoader first, then falls back to the app’s own resources_dir. Standalone: reads resources_dir directly, with a small suffix-based mime map (octet-stream default).

Return type:

tuple[bytes, str] | None

make_callable(node, request)

Build the complete call this app wants to execute for node.

Not just the kwargs: the whole call, arguments AND lifecycle — packaged in the vehicle the handler needs. The node declares its handler’s nature (genro-routes marks async entries so iscoroutinefunction(node) is honest); the caller runs the result via smartasync(...)() either way:

• sync handler -> sync zero-arg callable -> executor thread, which is where thread-local resource teardown belongs (a thread-local db connection must be closed on the same thread it opened, which the request-level cleanup on the loop cannot do);

• async handler -> async zero-arg callable awaited on the event loop, with route_cleanup running there AFTER the handler — an async handler must not lean on thread-local resources.

Two hooks vary the behaviour, this wrapper does not: bind_kwargs decides the arguments (REST spread, etc.); route_cleanup runs after the handler. Both default to a no-op-ish base; subclasses/mixins override the one they own.

Return type:

Callable[[], Any]

monitor_descriptor()

How this app’s panel LOOKS on the server monitor: a static descriptor.

The presentation complement of monitor_state(): the state carries the data (polled), the descriptor tells the monitor shell which panel renders it (fetched once — it is a per-class constant). panel names a renderer in the shell’s registry; this default (and any name the shell does not know) falls back to the generic panel — raw state as key/value rows and tables. Subclasses override to declare their specific panel and the extra endpoints it needs (see SpaMultiWorkerApplication).

Return type:

dict[str, Any]

monitor_state()

This app’s contribution to the server monitor: identity facts.

The server-level monitor aggregates one entry per mounted app by calling this on each. Subclasses override to add their specific panel data (registries, pools, gauges) on top of these base facts.

Return type:

dict[str, Any]

property mount_name: str

Return the name under which this app is mounted on the server.

on_init(**kwargs)

Called after base initialization. Override for custom setup.

Parameters:

**kwargs (Any) – Constructor kwargs from the app/application entry in the config.py recipe.

Return type:

None

on_route_resolved(node, request)

Hook called right after route resolution, before the handler runs.

A no-op seam on the base (FIXED): subclasses observe which route a request resolved to — e.g. a pool worker stamps per-connection activity here for routes not marked as system traffic (@route(meta_sysrpc=True)).

Return type:

None

on_shutdown()

Called when server stops. Override for custom cleanup.

Can be sync or async. Called in reverse order of startup.

Return type:

None

on_startup()

Called when server starts. Override for custom initialization.

Can be sync or async. Called after all apps are mounted.

Return type:

None

openapi_info: ClassVar[dict[str, Any]] = {}

path_in_app(path)

Return the request path relative to this app’s mount.

The server demultiplexes on the first path segment; an app resolves the remainder in its own router. The empty mount keeps the whole path.

Return type:

str

property resources_dir: Path | None

Return path to app’s resources directory.

Derived from app_path (the app’s own location), so an app finds its resources whether mounted on the server or used standalone.

rest_channel: ClassVar[str] = 'rest'

route_cleanup()

Teardown after the handler, in the executor thread. Base: nothing.

A subclass/mixin that owns a resource (e.g. a legacy db connection) overrides this to release it here — where it is thread-correct, unlike the request-level cleanup that runs on the loop.

Return type:

None

property server: AsgiServer | None

Return the server that mounted this app (semantic alias for _routing_parent).

spread_over_params(node, data)

Spread a dict of values over the handler’s declared parameters.

Shared by every wire dialect (REST body, MCP arguments): a structured payload is fitted to a scalar-parameter handler. Keeps data whole when the handler declares no signature (fields is None — no pydantic plugin) or accepts **kwargs; otherwise keeps only the declared names, dropping extras. An empty fields list means a known no-parameter handler: everything is dropped. Reads the neutral params block.

Return type:

dict[str, Any]

static(file='')

Serve a static file from the app’s resources directory.

Returns Path - set_result() handles mime type detection.

Raises:

HTTPNotFound – If no file is named, or the resource does not exist.

Return type:

Path

OpenApiApplication

Application that serves an OpenAPI schema and Swagger UI.

OpenApiApplication package.

class genro_asgi.applications.openapi_application.McpOpenApiApplication(**kwargs)

Bases: OpenApiApplication

OpenApiApplication that also exposes its API router as MCP tools.

mcp_name: str = 'genro-mcp'

mcp_name_segment: str = 'mcp'

mcp_version: str = '1.0.0'

on_init(routing_class=None, module=None, **kwargs)

Initialize; with no routing_class/module, direct mode (see module doc).

In direct mode the app’s own @route methods are the API: pydantic is plugged on the app router (signatures for schema and adaptation) and the channel plugin with default rest (a method joins the MCP face only via @route(channel_channels="mcp,...")). The attached MCP face inherits the pydantic plug from the app router.

Return type:

None

tool_separator: str = '.'

class genro_asgi.applications.openapi_application.OpenApiApplication(**kwargs)

Bases: AsgiApplication

Wrap a RoutingClass (or the app’s own @route methods) as REST + OpenAPI.

Adds meta endpoints under _meta/ (schema, docs, splash) alongside the API: - /{app}/_meta/schema_json — OpenAPI 3.1 JSON schema - /{app}/_meta/docs — Swagger UI page - /{app}/_meta/index — splash page

Two ways to supply the API: - mounted mode: pass routing_class= (or module=); the class is

attached under api_name at /{app}/{api_name}/endpoint.

• direct mode: subclass and write @route methods on the app itself; endpoints sit at the app root /{app}/endpoint.

Example (mounted):

app = OpenApiApplication(routing_class=api)
server.mount("sourcerer", app)

Example (direct):

class ShopApi(OpenApiApplication):
    @route(media_type="application/json")
    def items(self) -> dict: ...

property api_info: dict[str, Any]

the mounted class’s, else the app’s, else empty.

Type:

OpenAPI info dict

property api_name: str

Path segment the routing class is attached under.

bind_kwargs(node, request)

Reconcile a JSON body with a REST handler’s scalar parameters.

A JSON body arrives as a single body_data dict (request.py). A classic REST handler declares scalar parameters, not body_data, so the dict must be spread over the fields the handler accepts. The handler signature comes from the node’s neutral params block (never inspect).

The whole body_data is kept when the handler itself declares body_data or accepts **kwargs — it wants the raw object or any key. Otherwise the dict is spread over the declared names, extras dropped.

Return type:

dict[str, Any]

property docs_style: str

Documentation style — “swagger”, “redoc”, or “off”.

on_init(routing_class=None, module=None, docs='swagger', api_name='api', **kwargs)

Initialize with a routing class instance or import path.

Parameters:

• routing_class (RoutingClass | None) – RoutingClass instance to mount.

• module (str | None) – Import path “package.module:ClassName” (alternative to routing_class).

• docs (str) – Documentation style — “swagger”, “redoc”, or “off”.

• api_name (str) – Path segment the routing class is attached under.

• **kwargs (Any) – Passed to the routing class constructor if module is used.

Return type:

None

openapi_info: ClassVar[dict[str, Any]] = {}

McpApplication

MCP Streamable HTTP transport application.

McpApplication package.

class genro_asgi.applications.mcp_application.McpApplication(**kwargs)

Bases: AsgiApplication

MCP Streamable HTTP transport as a genro-asgi application.

Accepts an external genro-routes Router and exposes its @route entries as MCP tools via JSON-RPC 2.0 over HTTP POST.

Mounts inside the server like any other app and participates in the full middleware chain (auth, CORS, errors). The index route handles all JSON-RPC messages via the default_entry mechanism.

Usage:

class SourcererMcpApp(McpApplication):
    mcp_name = "sourcerer"
    mcp_version = "1.0.0"

    def on_init(self, **kwargs):
        api = SourcererAPI(...)
        self.set_router(api.route)

__init__(**kwargs)

Initialize with optional mcp_name/mcp_version/tool_separator overrides.

index()

JSON-RPC 2.0 endpoint for MCP Streamable HTTP.

Reads the parsed body from request.data (populated by the framework via asgi_data) and delegates dispatch to the MCP engine. Notifications (no “id”) get HTTP 202 and no body; requests get a JSON-RPC response.

Return type:

dict | None

mcp_name: str = 'genro-mcp'

mcp_version: str = '1.0.0'

set_router(router)

Set the genro-routes Router to expose as MCP tools.

The engine holds the JSON-RPC core (initialize / tools/list / tools/call); this app is the HTTP transport. Without a router, initialize still works and tools/list returns an empty list.

Return type:

None

tool_separator: str = '.'