Custom Routes in Modules
Flux0 lets you extend the API by adding your own HTTP endpoints from modules. This page shows two paths:
- Simple route — add a basic endpoint.
- Route with custom storage — initialize a store and use it in a handler.
All module routers are mounted under /api/modules/... during app startup.
Prerequisites
- Your module file must export lifecycle hooks:
init_module(container)andshutdown_module(). - Your module must export
get_routers(container) -> list[APIRouter]. - You can use common platform dependencies (e.g.,
AuthedUser) and services provided through the dependency injection.
Add a Simple Route
Implement get_routers(container) and return a FastAPI APIRouter with your local prefix and handlers.
from fastapi import APIRouter
from flux0_api.auth import AuthedUser
async def get_routers(container) -> list[APIRouter]:
router = APIRouter(prefix="/hello")
@router.get("/")
async def hello_world(user: AuthedUser):
return {"message": f"Hello {user.name} (id={user.id})!"}
return [router]
Effective path: GET /api/modules/hello/
The local prefix you set on the router (here
/hello) is appended after/api/modules.
Add a Route with Custom Storage
A store is a component that lets you persist and retrieve your own documents—such as user data, records, or other objects—using a backing database (e.g., NanoDB, MongoDB, etc.).
This example shows how to initialize a store (could be backed by a database such as NanoDB, MongoDB, etc.) and use it from a route.
A. Define the Store
from datetime import datetime, timezone
from typing import TypedDict
from flux0_core.ids import gen_id
class _StaticDocumentExample(TypedDict, total=False):
id: str
version: str
created_at: datetime
class ExampleStaticDocumentStore:
VERSION = "0.0.0"
def __init__(self, db):
self.db = db
self._col = None
async def setup(self):
# Create or open a collection/table in your database
self._col = await self.db.create_collection("example_static", _StaticDocumentExample)
async def create(self) -> _StaticDocumentExample:
created_at = datetime.now(timezone.utc)
doc = _StaticDocumentExample(
id=gen_id(),
version=self.VERSION,
created_at=created_at,
)
await self._col.insert_one(document=doc)
return doc
B. Initialize the Store in the Module
Use init_module to construct the store with the provided database, run setup(), and bind the instance back into the container.
from fastapi import APIRouter
from flux0_api.auth import AuthedUser
from examples.static.static_store import ExampleStaticDocumentStore
from .static_agent import StaticAgentRunner
async def init_module(container) -> None:
container[StaticAgentRunner] = StaticAgentRunner
example_store = ExampleStaticDocumentStore(container["db"])
await example_store.setup()
container[ExampleStaticDocumentStore] = example_store
async def shutdown_module() -> None:
print("Shutdown!")
C. Define Routes That Use the Store
Retrieve the bound store instance from the container inside your route handler.
async def get_routers(container) -> list[APIRouter]:
router = APIRouter(prefix="/static")
@router.get("/hello")
async def hello_world(user: AuthedUser):
example_store: ExampleStaticDocumentStore = container[ExampleStaticDocumentStore]
doc = await example_store.create()
return {
"message": f"Hello {user.name}. Your id is {user.id}. We created a static doc for you: {doc}"
}
return [router]
Effective path: GET /api/modules/static/hello
Behavior:
- Inserts a new document into the
example_staticcollection viaExampleStaticDocumentStore.create(). - Returns a message containing the user’s name, id, and the created document payload.
Notes & Tips
- Auth & middleware: Platform middlewares (authentication/authorization, logging, CORS, rate limits) apply to module routes automatically.
- Container: Prefer binding long-lived services in
init_moduleand retrieving them in handlers; this avoids repeated construction. - Prefixes: Keep router prefixes short and descriptive (e.g.,
/static,/my-feature). They’ll be appended after/api/modules. - Testing quickly: After your app starts, call your endpoint (e.g.,
GET /api/modules/static/hello) with a valid session token to verify both auth and storage.
Summary
- Your module must export
init_module(container)andshutdown_module(). - Return one or more routers from
get_routers(container). - Routes are available under
/api/modules/..., combining your router prefix and paths.
That’s it — your module can now expose new API endpoints without changing Flux0 core.