Skip to main content

Describe the domain.
Serve it over HTTP and MCP.

Fluvius is a Python 3.12+ framework for domain-driven backends. You write commands, aggregates, and queries. Fluvius records every change in an event log and serves the domain through FastAPI and MCP.

python -m pip install fluvius
Read the quick startBrowse the docs

Python 3.12+FastAPI · MCP

event log · user-profileappend-only
  1. 001030user-updatedupdate-useruser:2d9e0a47
  2. 001029user-invalidatedinvalidate-useruser:7f3a91c2
  3. 001028user-createdcreate-useruser:2d9e0a47
  4. 001027user-updatedupdate-useruser:c04e5b18
  5. 001026user-createdcreate-useruser:c04e5b18
  6. 001025user-updatedupdate-useruser:7f3a91c2
  7. 001024user-createdcreate-useruser:7f3a91c2

fluvius.fastapi · fluvius.mcp

Compose the app, one pipe at a time

Configurators take the app and return it, so ordering is explicit. Switch a line off to see what the app stops serving.

Read the docs
app/main.py
app = ()
The app serves
  • HTTP/api/docsOpenAPI UI
  • HTTP/healthFramework health and readiness
  • HTTP/api/authKeycloak / OpenID Connect sign-in, sessions, CSRF
  • HTTP/api/user-profile/create-user:post/userCommand endpoint
  • HTTP/api/user-profile/update-user:exec/user/{id}Command endpoint
  • HTTP/api/user-profile/invalidate-user:exec/user/{id}Command endpoint

6 HTTP surfaces · 0 MCP endpoint

fluvius.domain · fluvius.data · fluvius.query

Build a domain in six files

The user-profile example from the docs, in the order you would write it.

Read the docs
user_domain/model.py
import sqlalchemy as sa
from fluvius.data import SqlaDataSchema, SqlaDriver, DataAccessManager
from sqlalchemy.dialects import postgresql as pg
from fluvius.data import UUID_GENR


class UserConnector(SqlaDriver):
    __db_dsn__ = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"




class User(UserConnector.__data_schema_base__):
    __tablename__ = "user"


    _id = sa.Column(pg.UUID, primary_key=True, default=UUID_GENR)
    name = sa.Column(sa.String)
    _created = sa.Column(sa.DateTime(timezone=True))
    _updated = sa.Column(sa.DateTime(timezone=True))
    _etag = sa.Column(sa.String)
    _deleted = sa.Column(sa.DateTime(timezone=True))
    _creator = sa.Column(pg.UUID)
    _updater = sa.Column(pg.UUID)
    _realm = sa.Column(sa.String)




class FluviusAccessManager(DataAccessManager):
    __connector__ = UserConnector
    __automodel__ = True

fluvius.domain

What happens to a command

A call becomes a command, the aggregate applies it, and the result is appended to the event log. Scrub the log to replay one user.

Read the docs
  1. CallHTTP · MCPA request arrives through FastAPI or an MCP tool call.
  2. Commandcreate-userCreateUserCmd validates the payload against its Data model.
  3. AggregateUserAggregate.create_userThe @action decorator names the event and applies the change.
  4. Event loguser-createdSQLDomainLogStore appends the event to the domain log.
  5. QueryUserQueryThe read model serves the new state through typed fields.
Replay user:7f3a91c2seq 001036 · 4 of 4 events
_id
7f3a91c2
name
"Ada King"
_created
09:41:07
_updated
10:15:40
_deleted
10:15:40

The event log is the history. Replay it to see the state at any point.

What ships in the box

The parts most backends rebuild by hand, already wired to the same domain.

Public API reference

Access

Sign-in
Keycloak / OpenID Connect routes, sessions, CSRF protection
fluvius.fastapi
Authorization
roles_required on commands; Casbin policies when policy_required is set
fluvius.casbin

Operations

Errors
application/problem+json responses and request IDs
fluvius.fastapi
Health
/health and readiness state
fluvius.fastapi

History

Event log
SQLDomainLogStore records the events every command emits
fluvius.domain
Persistence
SQLAlchemy; PostgreSQL in production, SQLite for focused tests
fluvius.data

Background and realtime

Tasks and cron
Taskiq on PostgreSQL, Redis, or NATS brokers
fluvius.workq
Transport
NATS and MQTT, realtime bridge, JSON or CBOR codecs
fluvius.transport

Agents

MCP
Streamable HTTP tools; short-lived JWT, 900 s by default
fluvius.mcp

Install the core. Add packages when you need them.

Package map

Optional dependency groups

  • python -m pip install "fluvius[workq]"Redis, NATS, and PostgreSQL Taskiq brokers
  • python -m pip install "fluvius[transport]"MQTT and NATS transport clients
  • python -m pip install "fluvius[hatchet]"Hatchet workflow integration

Five minutes to a running API

  1. 1Installpython -m pip install fluvius
  2. 2Compose the appapp = create_app(pkginfo) | configure_routes()
  3. 3Run ituvicorn myapp.main:app --reloadOpen /docs for OpenAPI and /health for framework health.

Start with the quick start.

Read the quick startGitHub