LogalDeveloper/Owlbot

Fork 0

Table of Contents

Modules - Routes

@on_route Decorator

Parameters
Streaming Routes

Namespacing
Path Parameters

Single Segment
Custom Regex
Catch-All
Match Order

Response Types
Error Handling
URL Building
Protected Routes

Sending Session URLs

Templates
Request Access
Dynamic Registration

Available Methods

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.

@on_route Decorator

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"}

Parameters

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.
`requires_session`	`bool`	`False`	If `True`, the request must come from a browser connected to an Owncast user.
`requires_authenticated`	`bool`	`False`	If `True`, the connected Owncast user must be authenticated. Implies `requires_session`.
`requires_moderator`	`bool`	`False`	If `True`, the connected Owncast user must have the MODERATOR scope. Implies `requires_session`.

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.

Streaming Routes

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.

Namespacing

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.

Path Parameters

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 {}.

Single Segment

{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.

Custom Regex

{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)}

Catch-All

{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/x but not /api.

Match Order

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.

Response Types

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 automatically

Error Handling

Raised 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.

URL Building

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.

Protected Routes

Routes that need to know which chat user is visiting can require an Owlbot browser session. A browser session is created when a viewer runs the built-in !connect command in chat and opens the private link, or when your module sends them a session URL with ctx.session_url_for() from a command or user-bearing event handler.

requires_session=True only requires a linked browser session. Use requires_authenticated=True when the page should only be available to logged-in Owncast users. Use requires_moderator=True when it should only be available to moderators.

from aiohttp import web
from markupsafe import escape
from owlbot.api import RouteContext, on_route

@on_route("/panel", requires_moderator=True)
async def moderator_panel(ctx: RouteContext) -> web.Response:
    assert ctx.session is not None
    # Display names are user-controlled. Escape them directly or render
    # through Jinja templates so autoescape handles it.
    name = escape(ctx.session.user.display_name)
    return web.Response(
        text=f"<h1>Moderator panel for {name}</h1>",
        content_type="text/html",
    )

If a request does not have a usable session, Owlbot returns a connection page explaining that the viewer should run !connect in chat. The handler is not called. If a session exists but fails the authentication or moderator check, Owlbot returns a permission page instead.

When a protected route runs, ctx.session contains the linked Owncast user snapshot:

Property	Description
`ctx.session.user`	The Owncast `User` tied to this browser session.
`ctx.session.is_authenticated`	Whether that user was authenticated in the latest user snapshot Owlbot has seen.
`ctx.session.is_moderator`	Whether that user had the MODERATOR scope in the latest user snapshot Owlbot has seen.
`ctx.session.expires_at`	When the browser session expires.

Sending Session URLs

Call ctx.session_url_for(path) from a command handler or from an event handler whose event has a user field when you need to link to a protected page. Use it instead of ctx.routes.url_for(path) so Owlbot can link the browser session to that Owncast user. It accepts the same module-relative paths as url_for(), but returns a one-time /owlbot/connect/<token> URL for that user. When the user opens that link, Owlbot creates their browser session and redirects them to the route path you provided.

For example:

from owlbot.api import CommandContext, on_command

@on_command("panel", requires_moderator=True)
async def send_panel_link(ctx: CommandContext) -> None:
    url = ctx.session_url_for("/panel")
    await ctx.owncast_client.send_system_message_to_client(
        ctx.chat_event.client_id,
        f"Open your moderator panel: {url}",
    )

The path is module-relative ("/panel" or "panel"), not a full /owlbot/... path.

Warning: Owncast does not send webhook events when a user's authentication or moderator status changes. Owlbot updates session user data when it sees a later event for that user, and sessions expire after one hour by default. This means permission changes may take up to the session lifetime, or the next user event, to affect already-connected browsers.

Templates

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.

Request Access

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

Dynamic Registration

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,
    )

Available Methods

Method	Description
`ctx.routes.register(path, handler, methods=..., streaming=..., requires_session=..., requires_authenticated=..., requires_moderator=...)`	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.
`requires_session`	`bool`	Whether the route requires a connected browser session.
`requires_authenticated`	`bool`	Whether the route requires an authenticated Owncast user.
`requires_moderator`	`bool`	Whether the route requires the MODERATOR scope.

Built-in Modules

Creating New Modules