DEV Community: Vadim Kozyrevskii

Python CQRS: if we were writing our own coding agent

Vadim Kozyrevskii — Mon, 21 Sep 2026 12:32:47 +0000

Orders, sagas, outbox — I already did that. Everybody is writing agents now, the topic is warm, so I figured we might as well try the package here too. Not a product. Not a teardown of Claude. A stand, to see whether the handlers are enough.

If you are in a hurry: I made up an agent in an IDE. You tell it “find the auth bug and open a PR,” it reads the repo, patches, runs tests. Eight things it has to do, and how python-cqrs covers them. The model proposes ApplyPatch. The mediator runs it. It should not reach git push. It will happily retry a commit without being asked.

Links: GitHub · Docs · PyPI · Part 1

Why bother with an agent

Hi, still Vadim.

I already ran python-cqrs on orders and an outbox. The feed now looks like you should be ashamed if your résumé has no agent on it, and pretending I find that boring would be a lie. The topic is hot. Why not see whether the package still holds if the caller is a model with tools instead of FastAPI. I am not shipping a product. I am not turning anyone’s Claude inside out. I just want to know if the stuff we already write for services falls over.

For the article I made up a small agent in an IDE. You type “find the auth bug and open a PR.” It walks the repo, patches files, runs tests. So the rest of the text has something to hang on, here is what it has to do — I will come back to each item with code.

Read the repo: open files and search the code without touching the tree.
Patch files and run shell so a model retry does not do it twice.
Check permissions before a write: path inside the workspace, a deny-list, a human approval.
Stream the turn to a terminal or IDE: thought, tool call, slice of a file.
After a successful patch, run format, lint, and audit ourselves, without hoping the model remembers a second tool.
Apply the diff and run tests as one scenario: if it is red or the process dies in the middle, revert the files.
When a PR actually exists, tell CI and Slack for real.
If the LLM vendor or bash dies, stop honestly instead of spinning “I am thinking.”

I will start with the most harmless thing. Just looking at a file.

Tools as requests

Items 1 and 2 both look like tools from the outside. Almost everything an agent like ours does looks that way: the model pokes functions, and later you wonder why rm showed up twice.

Inside python-cqrs it is still a request to the mediator. The fork is simple. You can read and search forever — the tree does not change, that is a query, a model retry is almost none of our business. Patch and shell already change the world. So they are commands, and without a client_call_id a dropped stream will cheerfully spawn a second rm. You can title the postmortem “the model meant well.”

From the model, an MCP client, or your test, the door is the same. mediator.send(...). I like that because the adapter stays thin and the policy does not live in the prompt.

class ReadFile(cqrs.Request):
    path: str
    offset: int | None = None
    limit: int | None = None


class FileView(cqrs.Response):
    path: str
    content: str


class ReadFileHandler(cqrs.RequestHandler[ReadFile, FileView]):
    def __init__(self, workspace: "WorkspacePort") -> None:
        self._workspace = workspace

    async def handle(self, request: ReadFile) -> FileView:
        text = await self._workspace.read(request.path, request.offset, request.limit)
        return FileView(path=request.path, content=text)


class ApplyPatch(cqrs.Request):
    path: str
    diff: str
    client_call_id: str  # so a model retry does not apply the diff twice


class ApplyPatchResult(cqrs.Response):
    path: str
    changed: bool

You do not need to romanticize the SDK adapter after that. Tool schemas, these classes, send, model_dump() back as the tool result. Generate one tool per POSIX call and the model will drown in the catalog before it finds the auth bug. MCP servers already did that after wrapping GraphQL one-to-one, then wondered why tool pick fell off a cliff.

Fine. We can read a file. The model will also want to write — and it should not get a backdoor around your rules.

Permissions before anyone touches a file

Item 3 is “are we even allowed to write,” and I cover it with Chain of Responsibility. In the first article the same chain picked a payment method, which was routing. The mechanism here is the same, the job is different: until a link says yes, real ApplyPatch from item 2 should not see the request. First I check that the path sits inside the workspace. Then whether the shell looks like curl | bash. Then, if the file is sensitive, I ask a human. A system prompt that says “do not delete production” does not participate, and that is the point. A prompt negotiates. A chain refuses.

class InsideWorkspace(CORRequestHandler[ApplyPatch, ApplyPatchResult]):
    def __init__(self, workspace: "WorkspacePort") -> None:
        self._workspace = workspace

    async def handle(self, request: ApplyPatch) -> ApplyPatchResult | None:
        if not self._workspace.contains(request.path):
            return ApplyPatchResult(path=request.path, changed=False)
        return await self.next(request)


class NeedsHumanApproval(CORRequestHandler[ApplyPatch, ApplyPatchResult]):
    def __init__(self, gates: "ApprovalPort") -> None:
        self._gates = gates

    async def handle(self, request: ApplyPatch) -> ApplyPatchResult | None:
        if await self._gates.must_ask(request.path):
            await self._gates.wait(request.client_call_id)
        return await self.next(request)


class DoApplyPatch(CORRequestHandler[ApplyPatch, ApplyPatchResult]):
    def __init__(self, workspace: "WorkspacePort") -> None:
        self._workspace = workspace

    async def handle(self, request: ApplyPatch) -> ApplyPatchResult | None:
        changed = await self._workspace.apply(request.path, request.diff, request.client_call_id)
        return ApplyPatchResult(path=request.path, changed=changed)


def commands_mapper(m: cqrs.RequestMap) -> None:
    m.bind(ApplyPatch, build_chain([InsideWorkspace, NeedsHumanApproval, DoApplyPatch]))

Approval in item 3 immediately grows a household detail: people go get coffee, an HTTP request will not survive that, and the decision still has to. Then it is fair to leave a saga waiting and pick it up with recovery when Allow finally gets clicked. Do not store the whole chat there. The saga only needs to know which step froze, not how the model phrased “I will carefully fix auth” for the twelfth time.

We put item 3 in front of the write, but without item 4 the person in the IDE still cannot see what the agent is doing right now. Otherwise it is a black box that sometimes writes to git.

The turn you actually watch

Item 4 sounds simple: in a terminal I want to watch the agent think — a thought, a tool call, a slice of a file, another thought. The package has StreamingRequestHandler for that, and I would give it one session turn, not the whole process lifetime. The model sits inside the handler as a port: it plans, and every tool from items 1–2 still goes through mediator.send. If streaming starts writing to disk itself, skipping the chain from item 3, Allow becomes decoration — the patch just slips through the hole.

class AgentTurn(cqrs.Request):
    session_id: str
    message: str


class AgentChunk(cqrs.Response):
    kind: str  # "thought" | "tool" | "text"
    text: str


class CodingAgentHandler(cqrs.StreamingRequestHandler[AgentTurn, AgentChunk]):
    def __init__(self, llm: "LlmPort", mediator: cqrs.RequestMediator) -> None:
        self._llm = llm
        self._mediator = mediator
        self._events: list[cqrs.Event] = []

    def clear_events(self) -> None:
        self._events.clear()

    async def handle(self, request: AgentTurn) -> typing.AsyncIterator[AgentChunk]:
        tools = ["ReadFile", "Grep", "ApplyPatch", "RunTests"]
        async for step in self._llm.plan(request.message, tools=tools):
            if step.kind == "tool":
                result = await self._mediator.send(step.request)
                self._events.append(ToolCalled(session_id=request.session_id, tool=type(step.request).__name__))
                yield AgentChunk(kind="tool", text=result.model_dump_json())
            else:
                yield AgentChunk(kind=step.kind, text=step.text)

I would cap turns and tokens in LlmPort.plan, because otherwise the agent “improves” the same diff forever and spends your quota on literary treatment of an import. Item 8 sits here too: if the vendor’s network dies, a fallback can say “retry the turn” instead of a spinner that never ends and looks like deep thought. I would take SSE wiring for item 4 from the existing example: inventing a protocol on top of what the mediator already streams usually ends in a homemade bicycle and a buffering bug.

We showed the turn from item 4. What remains is item 5: the boring work after the patch already landed — and not hoping the model will call format on its own.

Hooks the model should not have to remember

Item 5 is exactly that: after a patch lands you almost always want a formatter, a search index bump, and a note that the file changed. If each of those is a separate tool next to item 2, the model will forget one of them, and you will get “fixed” auth with dancing formatting. It is calmer if the command appends FilePatched to events after it works, and the mediator fans that out to subscribers. Lint and audit do not wait on each other, so parallel event handlers are fair here. max_concurrent_event_handlers just stops them eating every core while the agent cheerfully patches ten files in a row.

class FilePatched(cqrs.DomainEvent, frozen=True):
    path: str
    session_id: str


class ApplyPatchHandler(cqrs.RequestHandler[ApplyPatch, ApplyPatchResult]):
    def __init__(self, workspace: "WorkspacePort") -> None:
        self._workspace = workspace
        self._events: list[cqrs.Event] = []

    @property
    def events(self) -> list[cqrs.Event]:
        return self._events

    async def handle(self, request: ApplyPatch) -> ApplyPatchResult:
        changed = await self._workspace.apply(...)
        self._events.append(FilePatched(path=request.path, session_id="..."))
        return ApplyPatchResult(path=request.path, changed=changed)


class FormatOnPatch(cqrs.EventHandler[FilePatched]):
    async def handle(self, event: FilePatched) -> None:
        await format_file(event.path)


class AuditOnPatch(cqrs.EventHandler[FilePatched]):
    async def handle(self, event: FilePatched) -> None:
        await audit.append(event)

That event lasts as long as the process, which is enough for item 5. Do not put item 7 on it: “tell CI the agent touched auth” dies with the process on a redeploy, CI hears nothing, and the tree already has different code. Neighboring systems get an outbox, but it is fair to reach that through item 6 — a scenario where two steps either finish together or roll back.

Patch, tests, revert

Item 6 is the honest scenario for an agent like this: apply the diff, run pytest, put the files back if it is red. That is not a hook from item 5 and not a second tool “just in case” next to item 2. It is two steps on two ports, and compensation should be a real operation, not an apology in chat. There is no LLM in this file, and there should not be: it already said “fix auth.” After that it is the same as a button in the IDE, except the button does not retry itself every second.

class FixAndVerifyContext(SagaContext):
    session_id: str
    path: str
    diff: str
    client_call_id: str
    applied: bool = False
    tests_ok: bool = False


class ApplyPatchStep(SagaStepHandler[FixAndVerifyContext, Response]):
    async def act(self, context: FixAndVerifyContext) -> SagaStepResult:
        await self._workspace.apply(context.path, context.diff, context.client_call_id)
        context.applied = True
        return self._generate_step_result(Response())

    async def compensate(self, context: FixAndVerifyContext) -> None:
        if context.applied:
            await self._workspace.reverse_last(context.client_call_id)


class RunTestsStep(SagaStepHandler[FixAndVerifyContext, Response]):
    async def act(self, context: FixAndVerifyContext) -> SagaStepResult:
        context.tests_ok = await self._tests.pytest()
        if not context.tests_ok:
            raise TestsFailed()
        return self._generate_step_result(Response())

    async def compensate(self, context: FixAndVerifyContext) -> None:
        return

Without client_call_id on the ports, item 2 breaks again: apply will stack the same diff twice, because agents retry more than people and feel less shame about it. Saga recovery is item 6 itself: the process died between patch and tests, and you need to finish the scenario instead of staring at git status. Idempotency helps when the model is merely trying to be useful and hits the same button again.

Opening a PR uses the same skeleton as item 6 — commit, push, gh pr create — but there is no universal compensate. What you undo depends on your git manners: I would not put a force-push to a shared branch into compensation even in an article. Still do not shove the OAuth thread into the step context. Flags like applied and tests_ok are enough. Let the chat live somewhere else, or one day you will open storage and find a 40k-token novel.

When the PR actually exists, we have reached item 7: you want CI and Slack to hear about it. The domain event from item 5 is too weak for that: it dies with the process. So a NotificationEvent goes in the same transaction as the row in your session table, and then the outbox from the first article does the rest. Outside the process that is a normal EventMediator: same handlers, input already coming from the bus. A subagent in another process can listen for SessionDelegated and never see your chat — the fact that the session was delegated is enough. Item 8 for a saga is the same as for a turn: a fallback if a step or the vendor suddenly dies.

Where I would stop

If you split item 6 into two independent tools, ApplyPatch and RunTests, that the model is supposed to assemble into carefulness, you usually get a red tree and a retried patch. For me “fix and verify” is one command or one saga, because otherwise carefulness lives in the prompt again.

I also would not keep the session transcript in saga storage, even for item 6. You get tens of thousands of tokens of JSON inside a step, and then someone debugs it on a Sunday. Let the chat live separately. The saga can have a couple of flags.

I would not drop item 3 because the prompt forbids sudo. We already saw how that ends: the prompt negotiates, bash still finds a way.

And if the whole checklist collapses to “what is OAuth,” with no items 2–7, you do not need python-cqrs. That is already on the when not to use page, and the hype does not cancel it. A confident model does not glue a patch and a push into one transaction.

You will not ship a product from this text, and that was never the goal. The goal was to check the list: a query covers item 1, a command item 2, CoR item 3, streaming item 4, events item 5, a saga item 6, the outbox item 7, a fallback item 8. The same types you already use for ordinary services, only the caller is now a model. Part 1 kept HTTP thin. Here the agent stays thin. It proposes. The mediator does the rest, and on a hype cycle that somehow looks fresh, even though the package was written for it.

Star, issues, PRs: python-cqrs. If you are already building something of your own on Lang* and the only missing piece is that a patch should not outlive a failed test run, this is that layer, without having to pretend you invented an agent framework.

Python CQRS: Building distributed systems without the pain (Sagas, Outbox, Event-Driven)

Vadim Kozyrevskii — Thu, 23 Apr 2026 11:50:21 +0000

Building distributed systems in Python? Here is how python-cqrs tackles consistency with orchestrated sagas, the mediator pattern, and a transactional outbox—without preaching theory for ten pages first.

TL;DR

One thread through the code: place an order (command), read it back (query), then run a payment saga: close the invoice, credit the balance—without leaving half a payment applied.
After money is in, a transactional outbox carries a notification to other services (for example, “provision this VPS”) with at-least-once delivery to the bus; in-process domain events stay a different, in-memory contract (see the small table in part 3).
Transports and frameworks stay at the edge: nothing in a handler hard-codes HTTP, Kafka, or a CLI. More patterns (streaming, chain of responsibility, and the rest) live in the docs.

Links: GitHub · Docs · PyPI python-cqrs

A story in three crashes

A user places an order. Payment succeeds.
A downstream service is supposed to do the next step—but the process dies (OOM, redeploy, node loss).
The other side never heard about the state change, yet money or inventory already moved.

That is the boring side of distributed systems: not “microservices are trendy,” but consistency when a commit in one place does not magically align the rest of the world.

Hi, I am Vadim, tech ex-lead of the financial infrastructure team at Timeweb Cloud. This post is a short intro to the library we use under our Python microservices—no epic “how we got here.” If you are tired of reinventing transactional outbox, sagas, and mediator wiring, this may save you time.

python-cqrs is a framework for CQRS and event-driven architecture. It grew out of a fork of diator and has since become its own stack for reliable service design.

The same bootstrap and mediator ideas apply to other flows; below is one concrete path only.

The system we are pretending to build

We model a narrow slice of a cloud-style control plane: a customer orders a product by SKU (think VPS or add-on), the order sits in a pending state until the gateway says money arrived, we close the invoice and move the customer balance in one coordinated saga, then we must notify another team’s service to actually create the instance. The code is toy-sized; BillingPort, LedgerPort, and the outbox are placeholders for your real services and database.

Part 1 — Command and query: create an order, read it back

We start with a command to open the order and a query to read the dashboard row. Same RequestMediator and mediator.send for both; only the request/response types and the handler differ.

Idea: one bootstrap call maps all Request types to handlers. Callers (FastAPI, worker, or tests) only call mediator.send(...).

import di
import cqrs
from cqrs.requests import bootstrap

class PlaceOrderCommand(cqrs.Request):
    customer_id: str
    product_sku: str
    amount: float

class PlaceOrderResult(cqrs.Response):
    order_id: str
    status: str

class PlaceOrderHandler(cqrs.RequestHandler[PlaceOrderCommand, PlaceOrderResult]):
    async def handle(self, request: PlaceOrderCommand) -> PlaceOrderResult:
        return PlaceOrderResult(order_id="ord-42", status="PENDING_PAYMENT")

class GetOrderById(cqrs.Request):
    order_id: str

class OrderView(cqrs.Response):
    order_id: str
    status: str
    amount: float
    product_sku: str

class GetOrderHandler(cqrs.RequestHandler[GetOrderById, OrderView]):
    async def handle(self, request: GetOrderById) -> OrderView:
        return OrderView(
            order_id=request.order_id,
            status="PENDING_PAYMENT",
            amount=99.0,
            product_sku="vps.s",
        )

def commands_mapper(m: cqrs.RequestMap) -> None:
    m.bind(PlaceOrderCommand, PlaceOrderHandler)
    m.bind(GetOrderById, GetOrderHandler)

mediator = bootstrap.bootstrap(
    di_container=di.Container(),
    commands_mapper=commands_mapper,
)

More wiring: Bootstrap.

Part 2 — Saga: payment in → close invoice → update balance

When payment clears, you may need two effects in order: mark the invoice paid, then credit the balance (or the ledger). If the second step fails, the first has to be undone in business terms. That is an orchestrated saga: act in order, compensate in reverse. Details and storage: Saga.

import dataclasses
import uuid
import di
import cqrs
from cqrs.saga import bootstrap
from cqrs.saga.saga import Saga
from cqrs.saga.step import SagaStepHandler, SagaStepResult
from cqrs.saga.storage.memory import MemorySagaStorage
from cqrs.saga.models import SagaContext
from cqrs.response import Response

@dataclasses.dataclass
class AfterPaymentContext(SagaContext):
    order_id: str
    customer_id: str
    invoice_id: str
    amount: float
    payment_id: str
    capture_ref: str | None = None
    balance_tx_id: str | None = None

class BillingPort:
    async def close_invoice(self, invoice_id: str, payment_id: str) -> str:
        return "inv-cap-1"

    async def reopen_invoice(self, invoice_id: str) -> None:
        pass

class LedgerPort:
    async def credit_balance(self, customer_id: str, amount: float) -> str:
        return "led-1"

    async def reverse_credit(self, tx_id: str) -> None:
        pass

# Step 1 — invoice paid in billing; compensate reopens if a later step fails

class CloseInvoiceStep(SagaStepHandler[AfterPaymentContext, Response]):
    def __init__(self, billing: BillingPort) -> None:
        self._billing = billing

    async def act(self, context: AfterPaymentContext) -> SagaStepResult:
        context.capture_ref = await self._billing.close_invoice(
            context.invoice_id, context.payment_id
        )
        return self._generate_step_result(Response())

    async def compensate(self, context: AfterPaymentContext) -> None:
        if context.capture_ref:
            await self._billing.reopen_invoice(context.invoice_id)

# Step 2 — credit customer balance; compensate reverses the ledger line

class CreditBalanceStep(SagaStepHandler[AfterPaymentContext, Response]):
    def __init__(self, ledger: LedgerPort) -> None:
        self._ledger = ledger

    async def act(self, context: AfterPaymentContext) -> SagaStepResult:
        context.balance_tx_id = await self._ledger.credit_balance(
            context.customer_id, context.amount
        )
        return self._generate_step_result(Response())

    async def compensate(self, context: AfterPaymentContext) -> None:
        if context.balance_tx_id:
            await self._ledger.reverse_credit(context.balance_tx_id)

class PaymentAfterSaga(Saga[AfterPaymentContext]):
    steps = [CloseInvoiceStep, CreditBalanceStep]

def saga_mapper(m: cqrs.SagaMap) -> None:
    m.bind(AfterPaymentContext, PaymentAfterSaga)

saga_mediator = bootstrap.bootstrap(
    di_container=di.Container(),
    sagas_mapper=saga_mapper,
    saga_storage=MemorySagaStorage(),
)

async def run_payment_flow() -> None:
    ctx = AfterPaymentContext(
        order_id="ord-42",
        customer_id="c1",
        invoice_id="inv-7",
        amount=99.0,
        payment_id="pgw-100",
    )
    async for _ in saga_mediator.stream(context=ctx, saga_id=uuid.uuid4()):
        pass

In the same module you can wrap the same sagas_mapper and storage in functools.lru_cache and return it from a saga_mediator_factory() used with fastapi.Depends in part 4, so the HTTP layer only calls saga_mediator_factory() and does not duplicate bootstrap. For a JSON body, add a small Pydantic model with the same fields as AfterPaymentContext and map it to the dataclass in the route (or map fields manually).

If the process dies between steps, persisted saga state and recovery (see the docs) let you finish the flow or the compensation, instead of a silent half-state.

Part 3 — Outbox: after the invoice is paid, connect the product

A separate provisioning stack should not rely on a “publish to Kafka, then hope” after the database already says “paid.” The transactional outbox stores the integration event in the same database transaction as your business update; a publisher process reads the outbox and sends to the bus (at-least-once; consumers should be idempotent). Overview: Transactional Outbox.

Register the wire name and payload type once, then in the same UoW as the payment commit, insert a row for NotificationEvent[ServiceProvisioningPayload].

import cqrs
from pydantic import BaseModel

class ServiceProvisioningPayload(BaseModel, frozen=True):
    order_id: str
    customer_id: str
    product_sku: str
    invoice_id: str

cqrs.OutboxedEventMap.register(
    "invoice_paid.provision_service",
    cqrs.NotificationEvent[ServiceProvisioningPayload],
)
# Persisted in the same DB transaction as “invoice paid” + outbox row in the command path.

Publisher loop (separate task or process):

import asyncio
import cqrs
from cqrs.message_brokers import kafka
from cqrs.adapters import kafka as kafka_adapters

broker = kafka.KafkaMessageBroker(
    producer=kafka_adapters.kafka_producer_factory(dsn="localhost:9092"),
)
producer = cqrs.EventProducer(
    message_broker=broker,
    repository=outbox_repository,
)

# Separate loop: batch read → send → commit outbox state

async def publish_loop() -> None:
    async for events in producer.event_batch_generator():
        for event in events:
            await producer.send_message(event)
        await producer.repository.commit()
        await asyncio.sleep(10)

Consumer with FastStream: Kafka delivers a NotificationEvent, you hand it to EventMediator.send and reuse the same handler style as inside the app.

import functools
import di
import cqrs
import faststream
import pydantic
from cqrs.events import bootstrap
from faststream import kafka

class ServiceProvisioningPayload(pydantic.BaseModel):
    order_id: str
    customer_id: str
    product_sku: str
    invoice_id: str

class OnInvoicePaidStartProvisioning(
    cqrs.EventHandler[cqrs.NotificationEvent[ServiceProvisioningPayload]]
):
    async def handle(
        self, event: cqrs.NotificationEvent[ServiceProvisioningPayload],
    ) -> None:
        return

@functools.lru_cache(maxsize=1)
def integration_mediator() -> cqrs.EventMediator:
    def em(m: cqrs.EventMap) -> None:
        m.bind(
            cqrs.NotificationEvent[ServiceProvisioningPayload],
            OnInvoicePaidStartProvisioning,
        )

    return bootstrap.bootstrap(di_container=di.Container(), events_mapper=em)

br = kafka.KafkaBroker(bootstrap_servers=["localhost:9092"])
app = faststream.FastStream(br)

@br.subscriber("invoice_paid.provision_service", group_id="provisioning", auto_commit=False)
async def on_invoice_paid(
    body: cqrs.NotificationEvent[ServiceProvisioningPayload],
    msg: kafka.KafkaMessage,
    mediator: cqrs.EventMediator = faststream.Depends(integration_mediator),
):
    await mediator.send(body)
    await msg.ack()

If the HTTP process uses bootstrap with a message_broker, see Event producing; the outbox remains the safe place for “this must be visible outside the monolith’s DB.”

In-process domain events vs outbox (same product, two channels)

	Domain event (in-process)	`NotificationEvent` + outbox
Use	Side effects in the same service in the same request (projection, email)	Other services must see the fact reliably
Delivery	In-process, at-most-once if the process dies	At-least-once to the bus after a successful outbox publish

Part 4 — FastAPI: presentation layer stays thin

The FastAPI integration guide recommends the same pattern: build mediators with bootstrap, inject them with fastapi.Depends, and keep route functions to HTTP only—no business rules in the module. Types, mediator_factory, saga_mediator_factory, and uuid live in your app module together with parts 1–3.

SagaMediator is the return type of cqrs.saga.bootstrap.bootstrap (see the Saga Mediator docs). AfterPaymentSagaRequest is the JSON body shape for the saga route; execution still uses AfterPaymentContext from part 2.

app = fastapi.FastAPI()

@app.post("/orders", status_code=201)
async def place_order(
    command: PlaceOrderCommand,
    mediator: cqrs.RequestMediator = fastapi.Depends(mediator_factory),
) -> PlaceOrderResult:
    return await mediator.send(command)

@app.get("/orders/{order_id}")
async def get_order(
    order_id: str,
    mediator: cqrs.RequestMediator = fastapi.Depends(mediator_factory),
) -> OrderView:
    return await mediator.send(GetOrderById(order_id=order_id))

@app.post("/invoices/close-after-payment", status_code=200)
async def close_invoice_after_payment(
    body: AfterPaymentSagaRequest,
    saga: cqrs.SagaMediator = fastapi.Depends(saga_mediator_factory),
):
    ctx = AfterPaymentContext(**body.model_dump())
    async for _ in saga.stream(context=ctx, saga_id=uuid.uuid4()):
        pass
    return {"status": "ok"}

POST/GET for orders follow the FastAPI docs (command + query + mediator.send). The close-invoice route only builds an AfterPaymentContext and drives the same async stream as in part 2: no steps or compensation logic in the handler. The outbox publisher and consumer from part 3 still run in separate processes; they are not called from these routes.

v5 heads-up (optional)

The docs describe upcoming breaking changes: Pydantic optional, default models moving toward dataclasses. Discussion #57 has migration notes—confirm on the repo you track.

Links and calls to action

Star the repo, open issues, or comment if this matches what you are building: python-cqrs on GitHub
Migrations and v5: Discussion #57
Deep dives: python-cqrs documentation

If you run Python microservices and live in the outbox / saga / CQRS space, what still hurts? I am curious—we might chip away at it together.