Edge of Context
Search

· Updated

Domain-Driven Design for AI Agents: Bounded Contexts, Tools, and Business Rules

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.

Domain-Driven Design (DDD) puts that business language and ownership at the center. For an agent, the model can interpret a request and propose a typed command, but an application service supplies trusted context and the domain model accepts or rejects the state change. This guide connects the vocabulary and boundary work to that execution path.

TL;DR. Use DDD when an agent changes business state in a domain with meaningful language, ownership, and rules. A schema validates the shape of a model proposal; the domain enforces its meaning. Do not equate agents with bounded contexts or generated JSON with a valid business decision.

The real problem is rule ownership

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:

Those questions are useful when a workflow is important enough to have policies and lifecycle. A simple read-only chatbot may not need aggregates, repositories, and events. Use DDD to manage domain complexity, not to decorate every LLM call.

Strategic design before code

Build a ubiquitous language

A ubiquitous language is vocabulary shared by domain experts and developers inside one bounded context. 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

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 word product modeled differently across bounded contexts

A bounded context owns its model and translates at its boundary. It is not automatically a microservice, repository, agent, or team, although those boundaries often align.

This distinction matters in agent design:

Start with a context map before drawing an agent graph. Otherwise the graph tends to reproduce tool availability rather than the business.

Classify the subdomains

DDD commonly separates:

For a task assistant, task management may be core, scheduling supporting, and notification delivery generic.

Task management, scheduling, and notifications as separate contexts

The classification guides investment. It does not mean every box needs an LLM.

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.

Aggregates and invariants

An aggregate is a consistency boundary. Its root exposes the operations that can change members and protects invariants such as:

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. An application service coordinates one use case: load state, invoke the domain operation, save with an expected version, and publish resulting events.

The domain should not call an LLM, HTTP client, or ORM. Those are adapters around the use case.

Domain events are facts, not a message bus

TaskAdded is a past-tense fact raised by the domain. The application can persist it in an outbox with the aggregate update, then publish an integration event after commit. Sending directly to a broker from an entity risks publishing an event for a transaction that later fails.

Events can coordinate agents, but they do not make coordination reliable by themselves. Delivery semantics, idempotency, ordering, and versioned contracts remain infrastructure work.

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 analogy is useful as long as validation and policy remain separate.

Model output translated into a domain command and checked by deterministic rules

The boundary has four steps:

  1. Constrain and parse: require a typed output contract.
  2. Normalize: resolve dates, units, identifiers, and locale using trusted context.
  3. Authorize: decide whether this actor may request the operation.
  4. Execute: invoke an aggregate method that enforces the invariant.

Pydantic can reject a missing field or invalid enum. 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 safely

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


class AddTaskProposal(BaseModel):
    description: str = Field(min_length=1, max_length=200)
    due_date: date | None = None
    priority: Literal["low", "normal", "high"] = "normal"

If 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

from dataclasses import dataclass, field
from datetime import date
from uuid import UUID, uuid4


@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: dict[UUID, Task] = field(default_factory=dict)
    _events: list[object] = field(default_factory=list)

    @property
    def tasks(self) -> tuple[Task, ...]:
        return tuple(self._tasks.values())

    def pull_events(self) -> tuple[object, ...]:
        events = tuple(self._events)
        self._events.clear()
        return events

    def add_task(
        self,
        description: str,
        due_date: date | None,
        priority: str,
    ) -> Task:
        normalized = " ".join(description.casefold().split())
        duplicate = any(
            " ".join(task.description.casefold().split()) == normalized
            and task.due_date == due_date
            for task in self._tasks.values()
        )
        if duplicate:
            raise ValueError("A matching task already exists for that date")

        task = Task(uuid4(), description.strip(), due_date, priority)
        self._tasks[task.task_id] = task
        self._events.append(TaskAdded(task.task_id, self.owner_id))
        return task

The example omits the TaskAdded definition for brevity. In a complete domain module it would be an immutable value object. The aggregate exposes a tuple view rather than its mutable dictionary, so callers cannot append around add_task().

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 repository port

from typing import Protocol
from uuid import UUID


class ConcurrentUpdate(Exception):
    pass


class TaskListRepository(Protocol):
    def get(self, owner_id: UUID) -> TaskList: ...

    def save(self, task_list: TaskList, expected_version: int) -> None: ...

The infrastructure adapter can implement optimistic concurrency with a version column. The domain contract states what matters without depending on SQLAlchemy or a particular database.

4. Coordinate the use case

from uuid import UUID


class AddTaskService:
    def __init__(
        self,
        repository: TaskListRepository,
        authorizer: TaskAuthorizer,
        outbox: Outbox,
    ) -> None:
        self.repository = repository
        self.authorizer = authorizer
        self.outbox = outbox

    def execute(
        self,
        actor_id: UUID,
        owner_id: UUID,
        proposal: AddTaskProposal,
    ) -> Task:
        self.authorizer.require_add_permission(actor_id, owner_id)

        task_list = self.repository.get(owner_id)
        expected_version = task_list.version
        task = task_list.add_task(
            description=proposal.description,
            due_date=proposal.due_date,
            priority=proposal.priority,
        )

        # Implement both writes in one database transaction.
        self.repository.save(task_list, expected_version)
        self.outbox.add_all(task_list.pull_events())
        return task

The comment about one transaction is essential: repository save plus outbox insert must succeed or fail together. A unit-of-work abstraction can own that transaction when the concrete repository and outbox share a database.

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.

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.

A tool result should distinguish failures the agent can act on: invalid proposal, unauthorized actor, domain conflict, concurrent update, and unavailable infrastructure. Do not flatten all failures into a string that invites blind retries.

Test the boundary in layers

Domain tests

Test aggregates without a model, network, or database:

Application tests

Use fake repositories and authorizers to verify loading, authorization order, expected-version saves, outbox behavior, and error mapping.

Model-contract evaluations

Evaluate the probabilistic adapter separately:

An end-to-end test should then confirm that bad proposals never bypass the same domain methods used by trusted interfaces.

When the design is working

You should be able to change the model provider without changing a domain test. A policy change should modify one aggregate or domain service rather than several prompts. A trace should use terms the owning team recognizes. A malformed or unauthorized proposal should fail before persistence, and a concurrent save should fail rather than silently overwrite state.

That is the practical value of DDD for agents. It does not make a model deterministic. It makes the system’s authority, language, and consistency boundaries explicit enough that the model does not need to be.

References