Quick Start

The smallest useful Fluvius application is a FastAPI app created from package metadata and extended with pipe-compatible configurators.

1. Add package metadata

myapp/pkginfo.py

__name__ = "myapp"
__title__ = "My Fluvius API"
__description__ = "A small Fluvius service"
__version__ = "0.1.0"
__build_no__ = 0
__build_ts__ = 0

2. Create the app

myapp/main.py

from fastapi import Request
from pipe import Pipe

from fluvius.fastapi import create_app

from . import pkginfo


@Pipe
def configure_routes(app):
    @app.get("/hello")
    async def hello(request: Request):
        return {"message": "Hello from Fluvius"}

    return app


app = create_app(pkginfo) | configure_routes()

Run it with an ASGI server:

uvicorn myapp.main:app --reload

Open /docs for OpenAPI and /health for framework health. When FLUVIUS_API_BASE is set, framework-owned routes are mounted beneath that base path.

3. Add framework capabilities

Configurators compose left to right:

from fluvius.fastapi import (
    configure_authentication,
    configure_domain_manager,
    configure_query_manager,
    create_app,
)
from fluvius.mcp import configure_mcp_server

app = (
    create_app(pkginfo)
    | configure_authentication(auth_profile_provider=MyProfileProvider)
    | configure_domain_manager(UserDomain)
    | configure_query_manager(UserQueryManager)
    | configure_mcp_server()
)

configure_domain_manager accepts domain classes as positional arguments; configure_query_manager accepts query-manager classes. MCP authentication is enabled by default and requires FLUVIUS_MCP_SECRET_KEY.

Where domain behavior lives

A domain binds an aggregate, state manager, log store, and registered command classes:

from fluvius.data import DataModel
from fluvius.domain import Domain, MessageDispatcher, SQLDomainLogStore


class UserDomain(Domain):
    __namespace__ = "user-profile"
    __aggregate__ = UserAggregate
    __statemgr__ = UserStateManager
    __logstore__ = SQLDomainLogStore
    __msgdispatcher__ = MessageDispatcher


class CreateUser(UserDomain.Command):
    class Meta:
        key = "create-user"
        name = "Create User"
        resource_init = True
        policy_required = False

    class Data(DataModel):
        name: str

    async def _process(self, aggregate, statemgr, payload):
        user = await aggregate.create_user(payload)
        yield aggregate.create_response({"_id": str(user._id)})

Commands are normally executed through mounted domain HTTP endpoints or a DomainWorkqClient, not by constructing an obsolete Sanic context directly.