Repository navigation
Modules Routes
Modules can register HTTP routes on the built-in aiohttp web server, which is shared with the webhook receiver. This is useful for web UIs, API endpoints, health checks, or anything else served over HTTP.
The @on_route decorator registers a function as a route handler. Basic usage:
from owlbot.api import RouteContext, on_route
@on_route("/stats")
async def stats(ctx: RouteContext) -> dict:
return {"viewers": 42, "uptime": "3h 12m"}| Parameter | Type | Default | Description |
|---|---|---|---|
path |
str |
(required) | URL path relative to the module's namespace. |
methods |
list[str] | None |
["GET"] |
HTTP methods to accept. |
streaming |
bool |
False |
If True, the handler runs without a timeout. Use for long-lived connections like SSE streams. The handler is responsible for its own keepalives, idle detection, and cleanup. |
A route that accepts multiple methods:
@on_route("/config", methods=["GET", "POST"])
async def config_page(ctx: RouteContext) -> dict:
if ctx.request.method == "POST":
data = await ctx.request.json()
# ... update something
return {"status": "updated"}
return {"current_config": ctx.config.as_dict()}Multiple handlers can be registered on the same path with different methods. This allows splitting logic across separate functions instead of branching on ctx.request.method:
@on_route("/data", methods=["GET"])
async def get_data(ctx: RouteContext) -> dict:
return {"items": [1, 2, 3]}
@on_route("/data", methods=["POST"])
async def create_data(ctx: RouteContext) -> dict:
body = await ctx.request.json()
# ... store the data
return {"created": True}Each handler's method set must not overlap with any other handler on the same path. Registering a handler whose methods conflict with an existing one raises ValueError.
Routes that hold a connection open for a long time (e.g. Server-Sent Events) should set streaming=True to bypass the handler timeout (default 30 seconds):
import asyncio
from aiohttp import web
from owlbot.api import RouteContext, on_route
@on_route("/events", streaming=True)
async def sse_handler(ctx: RouteContext) -> web.StreamResponse:
resp = web.StreamResponse()
resp.content_type = "text/event-stream"
await resp.prepare(ctx.request)
while True:
await resp.write(b"data: ping\n\n")
await asyncio.sleep(30)Streaming handlers manage their own lifecycle. Owlbot does not impose any timeout or keepalive, so a handler that never writes will run indefinitely. Write periodically to detect closed connections, as the next write after a client disconnects will raise ConnectionResetError.
Routes are automatically prefixed under /owlbot/<module_name>/. So if the module stats registers @on_route("/dashboard"), the actual URL is:
/owlbot/stats/dashboard
This avoids conflicts between modules and with the webhook endpoint. Only the relative path is needed in the decorator.
Route paths support {name} patterns for capturing dynamic segments, using the same syntax as aiohttp. When a request arrives, the URL path is matched against registered route patterns and the captured segments populate ctx.match_info.
Given the route @on_route("/users/{id}/posts/{slug}") and a request to /owlbot/my_module/users/42/posts/hello-world:
| Property | Value |
|---|---|
ctx.match_info["id"] |
"42" |
ctx.match_info["slug"] |
"hello-world" |
For plain routes without {...} patterns, ctx.match_info is an empty dict {}.
{name} matches a single path segment (no /):
@on_route("/users/{id}")
async def get_user(ctx: RouteContext) -> dict:
user_id = ctx.match_info["id"] # e.g., "123"
return {"user_id": user_id}Registered as /owlbot/my_module/users/{id}, this matches /owlbot/my_module/users/123 but not /owlbot/my_module/users/123/posts.
{name:regex} matches with a custom regex pattern:
@on_route("/users/{id:\\d+}")
async def get_user(ctx: RouteContext) -> dict:
user_id = ctx.match_info["id"] # only digits
return {"user_id": int(user_id)}{name:.*} matches the remaining path, including / characters. This is useful for prefix-style routing:
@on_route("/api/{path:.*}", methods=["GET", "POST"])
async def api_catchall(ctx: RouteContext) -> dict:
remaining = ctx.match_info["path"] # e.g., "users/123/posts"
return {"path": remaining}Note: The catch-all requires at least one character after the prefix.
/api/{path:.*}matches/api/xbut not/api.
Routes are matched in registration order (first match wins). Routes defined with @on_route at import time are registered before dynamically registered routes, so they naturally get priority.
Route handlers support four return types:
| Return Type | Behavior |
|---|---|
dict |
Automatically serialized to JSON with application/json content type. |
web.Response |
Returned as-is. Full control over status, headers, body. |
web.FileResponse |
Serves a file from disk. Sets content type, length, and caching headers automatically. |
None |
Returns a 204 No Content response. |
For redirects or explicit HTTP status responses, raise an aiohttp.web.HTTPException subclass such as web.HTTPFound, web.HTTPSeeOther, web.HTTPForbidden, or web.HTTPNotFound.
from pathlib import Path
from aiohttp import web
from owlbot.api import RouteContext, on_route
# JSON response:
@on_route("/api/data")
async def api_data(ctx: RouteContext) -> dict:
return {"items": [1, 2, 3]}
# Full control:
@on_route("/page")
async def html_page(ctx: RouteContext) -> web.Response:
return web.Response(text="<h1>Hello</h1>", content_type="text/html")
# Static file:
@on_route("/logo")
async def logo(ctx: RouteContext) -> web.FileResponse:
return web.FileResponse(Path(__file__).parent / "static" / "logo.png")
# No content:
@on_route("/webhook", methods=["POST"])
async def incoming_webhook(ctx: RouteContext) -> None:
data = await ctx.request.json()
ctx.logger.info(f"Received: {data}")
# Returns 204 automaticallyRaised aiohttp.web.HTTPException values are passed through unchanged. Use them for redirects and explicit HTTP error responses. Other uncaught exceptions are logged and the client receives an empty 500 response. The same applies if the handler exceeds the configured handler_timeout (default 30 seconds); it is cancelled and the client receives a 500. Streaming handlers (streaming=True) are exempt from the timeout.
To get the full public URL for a route, use ctx.routes.url_for():
url = ctx.routes.url_for("/list")
# -> "https://owlbot.example.com/owlbot/my_module/list"This uses the public_base_url from config. If not set, it falls back to owncast.url. Set public_base_url if Owlbot is hosted on a different domain than your Owncast instance.
Every module has access to a Jinja2 template environment via ctx.templates. Templates are loaded from the module's templates/ directory first, then from Owlbot's core templates (which includes a Bootstrap base.html). Autoescape is enabled by default.
Modules that use templates should be structured as a package so the template files can be bundled alongside the code:
modules/my_module/
__init__.py
templates/
list.html
# modules/my_module/__init__.py
from aiohttp import web
from owlbot.api import RouteContext, on_route
@on_route("/list")
async def list_page(ctx: RouteContext) -> web.Response:
items = ["apples", "oranges", "bananas"]
html = ctx.templates.render("list.html", items=items)
return web.Response(text=html, content_type="text/html")Templates can extend Owlbot's core base.html to inherit a Bootstrap layout:
{# modules/my_module/templates/list.html #}
{% extends "base.html" %}
{% block title %}My Items{% endblock %}
{% block content %}
<h1>Items</h1>
<ul>
{% for item in items %}
<li>{{ item }}</li>
{% endfor %}
</ul>
{% endblock %}The base.html template provides four blocks: title, head, content, and scripts.
For advanced use (custom filters, globals, etc.), access the underlying Jinja2 Environment via ctx.templates.env.
ctx.request is a standard aiohttp.web.Request. Some common patterns:
# Path parameters (from {name} patterns):
user_id = ctx.match_info["id"] # e.g., "123"
# Query parameters:
format = ctx.request.query.get("format", "json")
# JSON body:
data = await ctx.request.json()
# Form body:
data = await ctx.request.post()
# Path info:
path = ctx.request.path
# Headers:
auth = ctx.request.headers.get("Authorization")
# Raw body:
body = await ctx.request.read()Routes can be registered and unregistered at runtime through ctx.routes:
from owlbot.api import ModuleContext, RouteContext, on_setup
async def dynamic_handler(ctx: RouteContext) -> dict:
return {"dynamic": True}
@on_setup
async def setup(ctx: ModuleContext) -> None:
ctx.routes.register(
path="/dynamic",
handler=dynamic_handler,
)Dynamic registration also supports path patterns:
async def user_handler(ctx: RouteContext) -> dict:
return {"user_id": ctx.match_info["id"]}
@on_setup
async def setup(ctx: ModuleContext) -> None:
ctx.routes.register(
path="/users/{id}",
handler=user_handler,
)| Method | Description |
|---|---|
ctx.routes.register(path, handler, methods=..., streaming=...) |
Register a route. Accepts the same parameters as @on_route. Returns RouteInfo. |
ctx.routes.unregister(path) |
Remove all handlers at the path. Returns True if found. |
ctx.routes.unregister(path, method=...) |
Remove only the handler covering that method. Returns True if found. |
ctx.routes.get(path) |
Get all handlers at the path. Returns list[RouteInfo]. |
ctx.routes.get(path, method=...) |
Get the handler for a specific method. Returns RouteInfo | None. |
ctx.routes.exists(path) |
Check if any handler is registered at the path. |
ctx.routes.exists(path, method=...) |
Check if a handler for a specific method exists. |
ctx.routes.url_for(path) |
Build a full public URL for a route. |
ctx.routes.module_routes |
Property: list of all RouteInfo for the module. |
RouteInfo has these fields:
| Field | Type | Description |
|---|---|---|
path |
str |
Relative path (e.g., "/list"). |
full_path |
str |
Full namespaced path (e.g., "/owlbot/my_module/list"). |
methods |
frozenset[str] |
HTTP methods. |
handler |
RouteHandler |
The handler function. |
module_name |
str |
Module that owns this route. |
streaming |
bool |
Whether the handler bypasses the timeout. |
Built-in Modules
Creating New Modules