Open WebUI tool pattern: self-registering FastAPI routes (insert before SPAStaticFiles catch-all)

route_test.py

"""
title: Route Registration Reference
author: Adam Smith
author_url: https://adamsmith.as
description: Reference pattern for Open WebUI tools that register their own FastAPI endpoints. Demonstrates idempotent route injection before the SPA catch-all, version-stamped dedup, and lazy init.
required_open_webui_version: 0.4.0
version: 1.0.0
licence: MIT
requirements:
"""

# ✨ Open WebUI Tool Route Registration Pattern
#
# OWUI tools run as singleton instances inside a FastAPI process. They can
# register custom HTTP endpoints by accessing __request__.app. However:
#
#   1. Routes added via app.add_api_route() land AFTER the SPAStaticFiles
#      catch-all mount, so they never match. We must insert before it.
#   2. When a tool's code is updated, OWUI exec()s the new module and calls
#      Tools() again, but old routes persist (no cleanup hook). We must
#      strip stale routes before re-registering.
#   3. __init__ has no access to the FastAPI app. Registration must happen
#      lazily on the first tool method invocation.
#   4. Tool deletion orphans routes (no on_delete hook). Use a namespaced
#      path convention (/api/v1/x/{tool_id}/...) to identify orphans.
#
# This file is a self-contained reference. Copy the helpers into your tool.

from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# Configuration -- change TOOL_ID to match your tool's actual id in OWUI
# ---------------------------------------------------------------------------

TOOL_ID = "route_test"

# A version stamp. Bump this (or derive from content hash) so that route
# re-registration only happens when the tool code actually changes, not on
# every invocation.
TOOL_VERSION = "1.0.0"

# All custom routes live under this namespace.
ROUTE_PREFIX = f"/api/v1/x/{TOOL_ID}"


# ---------------------------------------------------------------------------
# Route registration helpers (copy these into your tool)
# ---------------------------------------------------------------------------

def _insert_route_before_spa(app, path: str, endpoint, methods: list[str] = ["GET"]):
    """
    Register a FastAPI route and reposition it before the SPAStaticFiles mount.

    OWUI's main.py mounts SPAStaticFiles at path="" as the last route. It
    catches all unmatched requests and serves the SvelteKit frontend. Any
    route added after it via app.add_api_route() will never be reached.

    This helper adds the route, then moves it just before the SPA mount.
    """
    app.add_api_route(path, endpoint, methods=methods)

    routes = app.router.routes
    new_route = None
    spa_idx = None
    for i, r in enumerate(routes):
        if hasattr(r, "path") and r.path == path:
            new_route = r
        if type(r).__name__ == "Mount" and getattr(r, "path", None) == "":
            spa_idx = i
    if new_route is not None and spa_idx is not None:
        routes.remove(new_route)
        routes.insert(spa_idx, new_route)


def _strip_tool_routes(app, prefix: str):
    """Remove all routes whose path starts with the given prefix."""
    app.router.routes = [
        r for r in app.router.routes
        if not (hasattr(r, "path") and r.path.startswith(prefix))
    ]


def _register_tool_routes(app):
    """
    Register all custom routes for this tool. Idempotent and version-aware.

    Uses app.state to track which version of routes is currently installed.
    If the version matches, registration is skipped. If it differs (tool code
    was updated), old routes are stripped and new ones are inserted.
    """
    version_key = f"__{TOOL_ID}_route_version__"
    current = getattr(app.state, version_key, None)
    if current == TOOL_VERSION:
        return  # already registered at this version

    # Strip any routes from a previous version of this tool
    _strip_tool_routes(app, ROUTE_PREFIX)

    # -- Define your endpoints here ----------------------------------

    from fastapi.responses import JSONResponse

    async def health():
        """Health check / proof that route registration is working."""
        return JSONResponse({
            "tool_id": TOOL_ID,
            "version": TOOL_VERSION,
            "status": "ok",
        })

    _insert_route_before_spa(app, f"{ROUTE_PREFIX}/health", health, methods=["GET"])

    # -- End endpoint definitions ------------------------------------

    setattr(app.state, version_key, TOOL_VERSION)


# ---------------------------------------------------------------------------
# The Tool
# ---------------------------------------------------------------------------

class Tools:
    class Valves(BaseModel):
        pass  # Add admin-configurable settings here

    class UserValves(BaseModel):
        pass  # Add per-user settings here

    def __init__(self):
        self.valves = self.Valves()
        self.citation = False
        # NOTE: We cannot register routes here because we don't have access
        # to the FastAPI app yet (__request__ is only available in tool methods).

    async def check_routes(
        self,
        __user__: dict,
        __request__=None,
    ) -> str:
        """
        Verify that this tool's custom HTTP endpoints are registered and
        reachable. Call this to diagnose routing issues.
        """
        if not __request__:
            return "ERROR: No request context."

        app = __request__.app
        _register_tool_routes(app)

        # Introspect to confirm
        registered = [
            {"path": r.path, "methods": sorted(r.methods) if hasattr(r, "methods") else None}
            for r in app.router.routes
            if hasattr(r, "path") and r.path.startswith(ROUTE_PREFIX)
        ]

        import json
        return (
            f"ROUTES_OK: {len(registered)} route(s) registered for {TOOL_ID} v{TOOL_VERSION}.\n"
            f"{json.dumps(registered, indent=2)}"
        )