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.
Top comments (0)