Domain-Driven Design for AI Agents: Contexts and Rules
This guide is for Python engineers building AI agents that change business data. You will learn how Domain-Driven Design (DDD) organizes business vocabulary and rules, and how to turn a model proposal into an authorized operation.
Agent projects become difficult to change when prompts, code, and business processes use different terms. Compliance asks for a “policy check,” while the implementation exposes process_data(). The vague name hides which rule is being applied, who owns it, and where a change belongs.
DDD puts that business language and ownership at the center. The language model interprets a request and proposes an operation with named, typed fields. Application code supplies trusted user information, checks permission, and invokes the business rules before saving a change.
Rule ownership is the problem
Agent systems often distribute one rule across a system prompt, a tool description, an API handler, and a database constraint. The copies drift. A refund limit changes, one prompt remains old, and a syntactically valid tool call reaches the wrong policy.
DDD starts by asking different questions:
- Which team owns the rule?
- What language do domain experts use for it?
- Within which boundary does that term have one meaning?
- Which state changes must remain consistent together?
Those questions are useful when a workflow is important enough to have policies and a lifecycle. A simple read-only chatbot may not need the code patterns described later in this article. Use the code patterns when an agent changes state that has rules; skip them for read-only or single-step features.
Ask these four questions in order. Each yes adds one pattern that later sections explain.
- Does the agent change business data? If not, use domain terms in tools and prompts and stop.
- Do rules decide whether a change is allowed? Add an aggregate and an application service.
- Does one term mean different things to different teams? Add bounded contexts and a context map.
- Must another service react to the change? Add a domain event and a transactional outbox.
Strategic design before code
Build a ubiquitous language
A bounded context is a part of the system where business terms and rules have one agreed meaning. Its ubiquitous language is the vocabulary shared by domain experts and developers. If support operations say RefundRequest, approval limit, and settlement, those terms should appear in requirements, code, tool contracts, and evaluations.
This is more than choosing descriptive method names. Terms need definitions and examples. Does “approved” mean a manager clicked a button, the payment processor accepted the transfer, or both? Ambiguity discovered in a glossary is cheaper than ambiguity discovered in an agent trace.
Draw bounded contexts around models and ownership
A domain model is the objects, data, and rules used to represent a business area; it is distinct from the language model. The same noun can mean different things in different contexts. “Product” may be a stock-keeping unit in Inventory, a priced line in Billing, and a delivery commitment in Order Management.
The figure compares one shared Product class with one Product model per context. In the lower half, each arrow is a mapping that code performs when an order line passes to another context.
Each context owns its domain model. When information passes to another context, code maps the terms and identifiers between them. Fowler describes this separation as a way to keep one model from spanning every meaning of a term (bounded contexts). It is not automatically a microservice, repository, agent, or team, although those boundaries often align.
In agent design this means:
- one context may use several model calls or specialized agents internally
- one agent that spans several contexts needs explicit translation and authority for each
- orchestration is an application concern; it does not erase domain ownership
The next figure draws these three cases. Compare the second and third columns: in the second, one agent translates terms for two contexts; in the third, an orchestrator only sends commands, and each context applies its own rules.
Start with a context map showing the business areas and how they exchange information, before drawing an agent graph. Otherwise the graph tends to reproduce tool availability rather than the business.
Classify the subdomains
DDD commonly separates:
- Core domain: the capability that creates differentiated value
- Supporting subdomain: necessary, business-specific work that is not the differentiator
- Generic subdomain: a solved capability such as identity or email delivery
For a task assistant, task management may be core, scheduling supporting, and notification delivery generic. The map places each context in its subdomain band. Scheduling supplies dates to Task Management, which reports each added task to Notifications through a TaskAdded event.
Spend modelling effort on the core domain and write its rules yourself. Use an existing service for generic subdomains, such as an email provider for notifications. The classification does not decide where an LLM goes.
Tactical patterns define the state boundary
Entities and value objects
An entity has identity and a lifecycle. A task remains the same task after its description changes. A value object is defined by its values and is usually immutable: an email address, money amount, or time window.
Agent state fits the same split. A ConversationSession with its message history is an entity, and so is each Task the agent plans. A recorded tool result, such as a ToolOutput with the call arguments and the response, is a value object: it does not change after it is stored. Once a Task is an object, it can refuse completion until its dependencies finish, and that rule lives in one method instead of several prompt templates.
Aggregates and invariants
An aggregate groups related objects whose changes must obey the same rules (Evans, Domain-Driven Design Reference). Other code changes those objects through one entry object, called the aggregate root.
In the example below, TaskList is the root and its Task objects are members. An invariant is a rule that must remain true after each operation. This list rejects a task with the same normalized description and due date as an existing task. Other domains have different invariants:
- a completed task cannot be completed again
- an owner cannot have duplicate open reminders for the same day
- a refund cannot exceed the remaining refundable amount
- a plan accepts at most 10 open tasks, even when the model proposes 50
An aggregate does not become safe merely because a Python list sits behind an add_task() method. External code must not receive a mutable reference that bypasses that method. Persistence also needs concurrency control, or two valid requests can violate an invariant when saved simultaneously.
Repositories and application services
A repository loads and saves aggregates without leaking database concerns into the domain (Fowler, Repository). An application service coordinates one use case: load state, invoke the domain operation, save with an expected version, and publish resulting events.
A unit of work groups the saves of one use case into one database transaction: it commits all of them or none (Percival and Gregory, Unit of Work). The application service opens it, makes its changes, and calls commit(). Leaving without calling commit() discards the uncommitted changes.
The domain should not call an LLM, HTTP client, or ORM. Those belong in adapters, the code that connects the application to outside systems.
Domain events are facts, not a message bus
TaskAdded records a fact: a task was added. The application can store this event in an outbox table in the same database transaction as the task change. The transaction saves both changes together or neither. A separate publisher sends the stored event to other services only after the transaction succeeds. This is the transactional-outbox pattern; an event object alone does not provide it (Richardson, Transactional Outbox). Sending directly to a broker from an entity risks publishing an event for a transaction that later fails.
The figure compares three runs: the entity sends the event to the broker and the transaction then rolls back; the outbox with a rollback; and the outbox with a commit.
Events can coordinate agents, but they do not make coordination reliable by themselves. Delivery semantics, idempotency, ordering, and versioned contracts remain infrastructure work.
Choose the publisher’s target by who consumes the event. When the listeners run in the same process, an in-process signal library such as blinker can call them after the commit. When other services consume the event, send it to a broker such as RabbitMQ, or to Redis Streams if the team already runs Redis. Do not use Redis Pub/Sub for these events: it delivers each message at most once, and a subscriber that is disconnected loses it (Redis Pub/Sub delivery semantics).
Treat model output as an untrusted proposal
An LLM integration resembles an anti-corruption layer: it translates an external, probabilistic representation into terms the domain understands. The translation layer checks only the shape and values of the proposal. Permission stays in the application service and business rules stay in the aggregate.
A port is an interface listing operations the application needs. The repository port lists how to load and save task lists; a database adapter implements those operations. The figure draws the four packages as rings and shows code dependencies, not execution order. The application uses the aggregate and repository port; the infrastructure adapter implements that port. The domain does not depend on either the database adapter or the model.
In a Python project, the four layers can be four packages: domain/, application/, infrastructure/, and interface/. Swapping the database then touches infrastructure/; changing the model provider touches interface/.
The next diagram runs the four calls from the worked example below through the five steps of the list under it. Each row shows where a call stops and what the tool returns.
Code between the model and the domain runs these steps:
- Receive the proposal: the model returns
AddTaskProposal; treat every value in it as untrusted. - Parse and normalize: validate it against the typed contract, then resolve dates, units, identifiers, and locale using trusted context.
- Authorize: decide whether this actor may request the operation.
- Enforce the invariant: invoke an aggregate method that checks the rule and raises
TaskAdded. - Commit together: save the task list and the outbox event in one transaction.
Three tools can make the model return the typed output and validate it (steps 1 and 2):
- Pydantic AI, when you want an agent framework. Set the agent’s output type to a Pydantic model; when the output fails validation, it sends the error back to the model and asks again.
- Instructor, when you keep your provider SDK and only need parsed Pydantic objects. It wraps the client, and
max_retriesre-asks the model with the validation error. - The provider’s own structured-output mode, such as OpenAI Structured Outputs, when you want the provider to constrain generation to the JSON schema.
Send a validation error back to the model once or twice. If the proposal still fails, ask the user instead of retrying.
Pydantic can reject a missing field or a priority outside low, normal, high through its typed validation model (Pydantic documentation). It cannot decide that “tomorrow” resolves to the correct date, that the user owns the task list, or that a similar task is already open.
Worked example: add a task
Consider a request to add “Send the invoice” to a task list. The actor is the signed-in requester (actor_id); the owner is the person whose list will change (owner_id). They may be different people, so the application checks permission for that actor and list before making the change.
The model adapter supplies the task details as AddTaskProposal. The application service checks permission, asks TaskList to add the task, and commits the change and its TaskAdded event through a unit of work. Save the seven Python blocks from this section through “Run the example”, in order, as task_example.py. They form one module for Python 3.12 with Pydantic 2; the last block runs it.
1. Define the model-facing proposal
Keep the proposal close to what the model can infer. Do not ask it to invent database IDs or trusted owner identifiers.
from datetime import date
from typing import Literal
from pydantic import BaseModel, Field, field_validator
class AddTaskProposal(BaseModel):
description: str = Field(min_length=1, max_length=200)
due_date: date | None = None
priority: Literal["low", "normal", "high"] = "normal"
@field_validator("description")
@classmethod
def description_must_contain_text(cls, value: str) -> str:
value = value.strip()
if not value:
raise ValueError("description must contain non-whitespace characters")
return valueIf the user says “tomorrow,” the application should give the model an explicit local date or resolve the relative expression with a tested date parser. Never use the inference server’s clock as implicit business context.
2. Put the invariant in the aggregate
TaskList.add_task() rejects blank descriptions and duplicate tasks before changing the list. Keeping those checks here makes them apply whether a request comes from an LLM or an ordinary form.
from dataclasses import dataclass, field
from datetime import date
from uuid import UUID, uuid4
class DomainError(Exception):
"""A business rule rejected the operation."""
class DuplicateTask(DomainError):
pass
@dataclass(frozen=True)
class TaskAdded:
task_id: UUID
owner_id: UUID
@dataclass(frozen=True)
class Task:
task_id: UUID
description: str
due_date: date | None
priority: str
@dataclass
class TaskList:
owner_id: UUID
version: int
_tasks: tuple[Task, ...] = ()
_events: list[TaskAdded] = field(default_factory=list)
@property
def tasks(self) -> tuple[Task, ...]:
return self._tasks
def pull_events(self) -> tuple[TaskAdded, ...]:
events = tuple(self._events)
self._events.clear()
return events
def add_task(
self,
description: str,
due_date: date | None,
priority: str,
) -> Task:
description = description.strip()
if not description:
raise DomainError("Task description must contain non-whitespace characters")
normalized = " ".join(description.casefold().split())
duplicate = any(
" ".join(task.description.casefold().split()) == normalized
and task.due_date == due_date
for task in self._tasks
)
if duplicate:
raise DuplicateTask("A matching task already exists for that date")
task = Task(uuid4(), description, due_date, priority)
self._tasks += (task,)
self._events.append(TaskAdded(task.task_id, self.owner_id))
return taskThe domain raises its own exception types. pydantic.ValidationError is a subclass of ValueError, so a tool that caught ValueError to report a domain conflict would also catch invalid proposals.
TaskAdded is a frozen dataclass: a value object that records the fact. The aggregate keeps tasks in an immutable tuple, so callers receive no mutable collection that they can append to or edit around add_task(). Its mutating methods replace that backing tuple only after enforcing the rule.
Duplicate detection here is deliberately simple. Real rules may need locale-aware normalization, recurrence semantics, or a database uniqueness constraint as a final race-safe backstop.
3. Define the ports
These Python protocols list what the application needs from storage and from permission checks. TaskListRepository loads and saves one task list. UnitOfWork holds that repository and the pending events, and commits both. TaskAuthorizer raises Unauthorized when the actor may not change the list.
from typing import Protocol, Self
from uuid import UUID
class ConcurrentUpdate(Exception):
pass
class Unauthorized(Exception):
pass
class TaskListRepository(Protocol):
def get(self, owner_id: UUID) -> TaskList: ...
def save(self, task_list: TaskList, expected_version: int) -> None: ...
class UnitOfWork(Protocol):
task_lists: TaskListRepository
def __enter__(self) -> Self: ...
def __exit__(self, *exc_info: object) -> None: ...
def add_events(self, events: tuple[TaskAdded, ...]) -> None: ...
def commit(self) -> None: ...
class TaskAuthorizer(Protocol):
def require_add_permission(self, actor_id: UUID, owner_id: UUID) -> None: ...The database adapter must save only if the stored version still matches expected_version, the version loaded earlier. A successful save increments that version. If another request has changed the list, it raises ConcurrentUpdate instead of overwriting that request’s work. This is optimistic concurrency control; SQLAlchemy’s version counter documentation shows the underlying comparison. The protocol states the requirement; a concrete adapter must implement the check and update together.
4. Coordinate the use case
AddTaskService puts permission, loading, the domain operation, and persistence in order. Its actor and owner identifiers come from trusted application context, not the model proposal.
from uuid import UUID
class AddTaskService:
def __init__(self, uow: UnitOfWork, authorizer: TaskAuthorizer) -> None:
self.uow = uow
self.authorizer = authorizer
def execute(
self,
actor_id: UUID,
owner_id: UUID,
proposal: AddTaskProposal,
) -> Task:
self.authorizer.require_add_permission(actor_id, owner_id)
with self.uow:
task_list = self.uow.task_lists.get(owner_id)
expected_version = task_list.version
task = task_list.add_task(
description=proposal.description,
due_date=proposal.due_date,
priority=proposal.priority,
)
self.uow.task_lists.save(task_list, expected_version)
self.uow.add_events(task_list.pull_events())
self.uow.commit()
return taskIf a line before commit() raises, the unit of work discards the uncommitted task list change and event. A connection failure or timeout during commit() can leave its outcome unknown: the database may have committed before its reply was lost. Leaving the block cannot undo that committed transaction (Amazon Builders’ Library, Making retries safe with idempotent APIs).
The model is absent from this service. One adapter may obtain AddTaskProposal from an LLM, another from an HTTP form, and tests can construct it directly. The business behavior remains identical.
5. Add in-memory adapters
These adapters let the example run without a database. InMemoryUnitOfWork.commit() checks every version first and only then writes the task lists and the events, so it saves both or neither. AllowList is a fake authorizer for tests and demos.
import copy
from uuid import UUID
class InMemoryDatabase:
def __init__(self) -> None:
self.task_lists: dict[UUID, TaskList] = {}
self.outbox: list[TaskAdded] = []
class InMemoryTaskLists:
def __init__(self, db: InMemoryDatabase) -> None:
self.db = db
self.staged: dict[UUID, tuple[TaskList, int]] = {}
def get(self, owner_id: UUID) -> TaskList:
stored = self.db.task_lists.get(owner_id, TaskList(owner_id, version=0))
return copy.deepcopy(stored)
def save(self, task_list: TaskList, expected_version: int) -> None:
self.staged[task_list.owner_id] = (task_list, expected_version)
class InMemoryUnitOfWork:
def __init__(self, db: InMemoryDatabase) -> None:
self.db = db
def __enter__(self) -> "InMemoryUnitOfWork":
self.task_lists = InMemoryTaskLists(self.db)
self.events: list[TaskAdded] = []
return self
def __exit__(self, *exc_info: object) -> None:
# Leaving the block without commit() discards every staged change.
self.task_lists.staged.clear()
self.events.clear()
def add_events(self, events: tuple[TaskAdded, ...]) -> None:
self.events.extend(events)
def commit(self) -> None:
staged = self.task_lists.staged
for owner_id, (_, expected) in staged.items():
stored = self.db.task_lists.get(owner_id)
if (stored.version if stored else 0) != expected:
raise ConcurrentUpdate(f"task list {owner_id} changed")
for owner_id, (task_list, expected) in staged.items():
task_list.version = expected + 1
self.db.task_lists[owner_id] = task_list
self.db.outbox.extend(self.events)
staged.clear()
self.events.clear()
class AllowList:
def __init__(self, allowed: set[tuple[UUID, UUID]]) -> None:
self.allowed = allowed
def require_add_permission(self, actor_id: UUID, owner_id: UUID) -> None:
if (actor_id, owner_id) not in self.allowed:
raise Unauthorized(f"{actor_id} may not add tasks for {owner_id}")A database adapter implements UnitOfWork with one SQL transaction; the in-memory version shows the contract.
Map tools to application commands
Agent tools should expose use cases, not database primitives. Prefer:
add_task(description, due_date, priority)
complete_task(task_id)
reschedule_task(task_id, due_date)over:
insert_row(table, values)
update_record(table, id, patch)The first set speaks the domain language and gives the application a place to authorize and enforce rules. The second lets the model describe arbitrary persistence mutations. The figure sends the same duplicate request through both designs: the database tool saves a second copy, and the domain-command tool refuses it.
Agent framework code, such as a LangGraph node or a Pydantic AI tool, belongs in the interface layer and calls application services, as add_task_tool() below does. If orchestration in the application layer uses a framework, put it behind a port the application owns. The domain never imports the framework.
A tool result should distinguish failures the agent can act on. Do not flatten all failures into a string that invites blind retries. Each failure has a different next action:
| Failure | Raised as | Tool returns | What the agent does |
|---|---|---|---|
| Invalid proposal | pydantic.ValidationError | invalid_proposal | Fix the arguments from the error and call again, once or twice |
| Unauthorized actor | Unauthorized | unauthorized | Stop and tell the user; do not retry |
| Domain conflict | DomainError, e.g. DuplicateTask | domain_conflict | Tell the user and ask how to proceed |
| Concurrent update | ConcurrentUpdate | concurrent_update | Call the tool again once; the service reloads the list |
| Unavailable infrastructure | ConnectionError, TimeoutError | unavailable | Check the outcome before retrying; use the conditions below |
Here, unavailable reports an infrastructure exception; it does not prove that nothing was saved. Retry with backoff only after the application confirms no write occurred, or when it atomically saves the write and its result under a request ID and returns that result on retries. Reuse the same ID and arguments within its retention period (Amazon Builders’ Library, Making retries safe with idempotent APIs). Otherwise, reconcile the outcome using the application’s records. If those records cannot resolve it, tell the user the outcome is unknown.
The demo rejects duplicate task content but does not implement request idempotency: a retry after a committed write returns domain_conflict rather than the original success.
This tool function wraps AddTaskService.execute() and returns those codes. The framework passes the actor and owner from the signed-in session, not from the model.
from uuid import UUID
from pydantic import ValidationError
def add_task_tool(
service: AddTaskService,
actor_id: UUID,
owner_id: UUID,
arguments: dict,
) -> dict:
try:
proposal = AddTaskProposal.model_validate(arguments)
task = service.execute(actor_id, owner_id, proposal)
except ValidationError as exc:
return {"error": "invalid_proposal", "detail": str(exc)}
except Unauthorized:
return {"error": "unauthorized"}
except DomainError as exc:
return {"error": "domain_conflict", "detail": str(exc)}
except ConcurrentUpdate:
return {"error": "concurrent_update"}
except (ConnectionError, TimeoutError):
return {"error": "unavailable"}
return {"ok": True, "task_id": str(task.task_id)}Run the example
The last block calls the tool four times: one valid task, one duplicate with different case and spacing, one priority outside the allowed values, and one actor without permission.
from uuid import uuid4
if __name__ == "__main__":
alice, bob = uuid4(), uuid4()
db = InMemoryDatabase()
service = AddTaskService(InMemoryUnitOfWork(db), AllowList({(alice, alice)}))
def call(actor, arguments):
result = add_task_tool(service, actor, alice, arguments)
print(result.get("error", "ok"))
invoice = {"description": "Send the invoice", "due_date": "2026-10-05"}
call(alice, invoice) # ok
call(alice, {**invoice, "description": "send the INVOICE"}) # domain_conflict
call(alice, {**invoice, "priority": "urgent"}) # invalid_proposal
call(bob, invoice) # unauthorized
stored = db.task_lists[alice]
print(f"version={stored.version} tasks={len(stored.tasks)} outbox={len(db.outbox)}")With Python 3.12 installed, run these commands in the directory containing task_example.py:
python3.12 -m venv .venv
.venv/bin/python -m pip install "pydantic>=2,<3"
.venv/bin/python task_example.pyThe module prints:
ok
domain_conflict
invalid_proposal
unauthorized
version=1 tasks=1 outbox=1Only the first call changed state. The stored list is at version 1 with one task, and the outbox holds one TaskAdded event.
Tests for each layer
Domain tests
Test aggregates with pytest, without a model, network, or database:
- duplicate tasks are rejected
- valid tasks raise the expected event
- exposed collections cannot mutate internal state
- transition rules hold over repeated operations
The second call in the run above already checks the first item through the tool. A domain test makes the same call on TaskList directly and expects DuplicateTask.
Application tests
Use fake repositories and authorizers to verify loading, authorization order, expected-version saves, outbox behavior, and error mapping. The in-memory unit of work above is that fake. Save the following block separately as test_task_example.py, beside task_example.py. It opens two units of work on the same data to simulate two requests that load the same version:
from uuid import uuid4
import pytest
from task_example import ConcurrentUpdate, InMemoryDatabase, InMemoryUnitOfWork
def test_stale_save_raises_concurrent_update():
db, owner = InMemoryDatabase(), uuid4()
first, second = InMemoryUnitOfWork(db), InMemoryUnitOfWork(db)
with first, second:
mine = first.task_lists.get(owner) # both requests load version 0
theirs = second.task_lists.get(owner)
mine.add_task("Call the bank", None, "normal")
first.task_lists.save(mine, expected_version=0)
first.commit() # stored version is now 1
theirs.add_task("Book travel", None, "normal")
second.task_lists.save(theirs, expected_version=0)
with pytest.raises(ConcurrentUpdate):
second.commit()
assert [t.description for t in db.task_lists[owner].tasks] == ["Call the bank"]Install pytest in the same environment and run the test:
.venv/bin/python -m pip install pytest
.venv/bin/python -m pytest -q test_task_example.pyModel-contract evaluations
Evaluate the probabilistic adapter separately:
- intent and field extraction accuracy
- relative-date resolution with supplied timezone context
- refusal or clarification when required information is missing
- resistance to prompt injection inside quoted task text
- rate of schema-valid but semantically unusable proposals
For tests that should not call a real model, Pydantic AI provides TestModel, which calls the agent’s tools and returns schema-valid output, and FunctionModel, which lets the test decide each response (Pydantic AI testing). Keep the evaluations above for the real model.
An end-to-end test should then confirm that bad proposals never bypass the same domain methods used by trusted interfaces.
Order of work for a new agent project
- Interview the domain experts and write the glossary, with a definition and an example for each term.
- Draw the context map. Start with one core context.
- Model the entities and value objects. Put each rule in the method that changes the state.
- Choose the aggregate roots and write down their invariants.
- Define the repository, unit-of-work, and authorizer ports. Use in-memory adapters until the domain tests pass.
- Raise domain events for changes other contexts need, and write them to an outbox in the same transaction.
- Define the model-facing proposal types and the parser that validates them.
- Expose each use case as a tool that calls an application service and maps its failures.
- Add orchestration last, sending commands between contexts.
When the design is working
You should be able to change the model provider without changing a domain test. A policy change should touch one aggregate or domain service instead of several prompts. Traces should use terms the owning team recognizes. Malformed or unauthorized proposals should fail before persistence, and a concurrent save should fail instead of silently overwriting state.
The model will still make wrong proposals; the aggregate rejects the ones that break a rule.
References
- Eric Evans, Domain-Driven Design Reference — strategic and tactical pattern definitions
- Martin Fowler, Bounded Context — why one model should not span every meaning of a term
- Martin Fowler, Repository — persistence collection abstraction
- Harry Percival and Bob Gregory, Architecture Patterns with Python: Unit of Work — the unit-of-work pattern in Python
- Chris Richardson, Transactional Outbox — publishing after a database transaction without losing events
- Pydantic documentation — typed proposal validation
- Pydantic AI output — structured output with validation retries
- Instructor retrying — validated Pydantic output on top of a provider SDK