Skip to content

Multi-Domain Applications

A large system usually has more than one bounded context, and in Protean each one is a separate Domain instance. Here is how to structure an application that way. For the conceptual foundation, see Bounded Contexts.

When to use multiple domains

Use separate domains when your application has:

  • Distinct vocabularies: "Customer" in billing means something different from "Customer" in shipping
  • Independent lifecycles: The catalog team deploys weekly; the billing team deploys monthly
  • Different infrastructure needs: Orders need PostgreSQL; analytics needs Elasticsearch
  • Organizational boundaries: Separate teams own separate contexts

If your entire application shares one ubiquitous language and one team maintains it, a single domain is simpler and sufficient.

Project structure

Organize each bounded context as a separate Python package with its own domain instance, configuration, and elements:

my_app/
├── identity/
│   ├── __init__.py          # identity_domain = Domain(name="Identity")
│   ├── domain.toml          # Identity-specific config
│   ├── customer.py          # Customer aggregate
│   ├── events.py            # CustomerRegistered, etc.
│   └── handlers.py          # Identity command/event handlers
├── catalogue/
│   ├── __init__.py          # catalogue_domain = Domain(name="Catalogue")
│   ├── domain.toml
│   ├── product.py           # Product aggregate
│   └── handlers.py
├── fulfillment/
│   ├── __init__.py          # fulfillment_domain = Domain(name="Fulfillment")
│   ├── domain.toml
│   ├── shipment.py          # Shipment aggregate
│   ├── subscribers.py       # Consumes events from other domains
│   └── handlers.py
└── api/
    └── app.py               # FastAPI app wiring all domains together

Each domain's __init__.py creates its own Domain instance:

# my_app/identity/__init__.py
from protean import Domain

identity_domain = Domain(name="Identity")

# my_app/catalogue/__init__.py
catalogue_domain = Domain(name="Catalogue")

# my_app/fulfillment/__init__.py
fulfillment_domain = Domain(name="Fulfillment")

Independent configuration

Each domain can have its own domain.toml with separate database, broker, and event store settings:

# my_app/identity/domain.toml
[databases.default]
provider = "postgresql"
database_uri = "postgresql://localhost/identity_db"

[brokers.default]
provider = "redis"
URI = "redis://localhost:6379/0"

# my_app/catalogue/domain.toml
[databases.default]
provider = "elasticsearch"
database_uri = { hosts = ["http://localhost:9200"] }

[brokers.default]
provider = "redis"
URI = "redis://localhost:6379/1"

Alternatively, pass configuration programmatically:

# fragment
identity_domain = Domain(
    name="Identity",
    config={
        "databases": {
            "default": {
                "provider": "postgresql",
                "database_uri": "postgresql://localhost/identity_db",
            }
        }
    },
)

Wiring domains in FastAPI

Use DomainContextMiddleware to route HTTP requests to the correct domain context based on URL prefix:

# my_app/api/app.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from protean import current_domain
from protean.integrations.fastapi import (
    DomainContextMiddleware,
    register_exception_handlers,
)

# Also import identity_domain, catalogue_domain and fulfillment_domain
# from my_app.identity, my_app.catalogue and my_app.fulfillment.


@asynccontextmanager
async def lifespan(app: FastAPI):
    initialized = []
    try:
        # Initialize all domains at startup
        for d in [identity_domain, catalogue_domain, fulfillment_domain]:
            d.init()
            initialized.append(d)
            with d.domain_context():
                d.setup_database()
        yield
    finally:
        # Close only the domains whose init() finished, in reverse order
        for d in reversed(initialized):
            d.close()


app = FastAPI(lifespan=lifespan)

app.add_middleware(
    DomainContextMiddleware,
    route_domain_map={
        "/customers": identity_domain,
        "/products": catalogue_domain,
        "/shipments": fulfillment_domain,
    },
)
register_exception_handlers(app)


@app.post("/customers", status_code=201)
async def register_customer(payload: dict) -> dict:
    return {"domain": current_domain.name}


@app.get("/products")
async def list_products() -> dict:
    return {"domain": current_domain.name}

Requests to /customers/... automatically activate identity_domain; requests to /products/... activate catalogue_domain, and so on. Requests that don't match any prefix (e.g. /health) pass through without a domain context.

Cross-domain communication

Domains communicate through events, never by importing each other's aggregates or calling each other's services directly. Protean supports two integration paths, and the right choice depends on the strategic relationship between the bounded contexts.

Choosing the right integration path

Co-located domains Distributed domains
Deployment Same repo, same process Separate services, independent deploys
Schema ownership You control both sides; changes are visible in one codebase Upstream can change independently
Integration mechanism register_external_event() with typed event classes Subscribers as anti-corruption layers (raw dict payloads)
Best for Process managers coordinating cross-domain workflows; event handlers syncing state across co-located contexts Independent services, external systems, webhook ingestion
DDD pattern Conformist / Published Language Anti-Corruption Layer

Tip

This is not an either-or choice at the project level. A single application can use both paths, register_external_event for tightly coordinated domains in the same repo, and subscribers for external systems you don't control.

Co-located domains: Registered external events

When multiple domains live in the same repository and share the same event store, use register_external_event() to give your domain typed access to another domain's events. This is especially useful for process managers that coordinate workflows spanning multiple bounded contexts.

from protean.core.event import BaseEvent
from protean.fields import Float, Identifier, String


# Define the external event class in your domain.
# This is YOUR domain's representation of the external event.
# It does not import from the other domain's package.
class PaymentReceived(BaseEvent):
    payment_id = Identifier(required=True)
    order_id = Identifier(required=True)
    amount = Float()


class StockReserved(BaseEvent):
    order_id = Identifier(required=True)
    inventory_item_id = Identifier(required=True)
    quantity = Float()


class StockUnavailable(BaseEvent):
    order_id = Identifier(required=True)
    inventory_item_id = Identifier(required=True)


# Register external events with their type strings
fulfillment_domain.register_external_event(
    PaymentReceived, "Billing.PaymentReceived.v1"
)
fulfillment_domain.register_external_event(StockReserved, "Inventory.StockReserved.v1")
fulfillment_domain.register_external_event(
    StockUnavailable, "Inventory.StockUnavailable.v1"
)

Now process managers and event handlers can use these typed events directly:

from protean import handle


# The fulfillment domain's own order and event
@fulfillment_domain.aggregate
class Order:
    status = String(default="new")


@fulfillment_domain.event(part_of=Order)
class OrderPlaced:
    order_id = Identifier(required=True)


@fulfillment_domain.process_manager(
    stream_categories=[
        "fulfillment::order",  # Own domain
        "billing::payment",  # External domain stream
        "inventory::inventory_item",  # External domain stream
    ]
)
class OrderFulfillmentPM:
    order_id = Identifier()
    status = String(default="new")

    @handle(OrderPlaced, start=True, correlate="order_id")
    def on_order_placed(self, event: OrderPlaced) -> None:
        self.order_id = event.order_id
        self.status = "awaiting_payment"

    @handle(PaymentReceived, correlate="order_id")
    def on_payment_received(self, event: PaymentReceived) -> None:
        if self.status != "awaiting_payment":
            return
        self.status = "awaiting_stock"

    @handle(StockReserved, correlate="order_id")
    def on_stock_reserved(self, event: StockReserved) -> None:
        if self.status != "awaiting_stock":
            return
        self.status = "completed"
        self.mark_as_complete()

    @handle(StockUnavailable, correlate="order_id", end=True)
    def on_stock_unavailable(self, event: StockUnavailable) -> None:
        self.status = "stock_unavailable"

on_stock_reserved completes the process only when the order is waiting for stock, so it calls mark_as_complete() inside its guard. end=True would complete the process even on the early return. on_stock_unavailable ends the process in any state, so it uses end=True.

Why this works for co-located domains:

  • You control both sides of the schema. Changes are visible in one codebase and can be coordinated in a single pull request.
  • Typed events give you IDE support, static analysis, and deserialization validation, no parsing raw dicts.
  • The PM subscribes to external streams directly, keeping the coordination logic concise and traceable.

Important: The external event class is defined in your domain, not imported from the other domain's package. You own the class; the type string ("Billing.PaymentReceived.v1") is the shared contract.

For syncing state across co-located contexts (without a process manager), you can also use event handlers with stream_category:

# Also import current_domain from protean.


# Fulfillment's own copy of the identity event, and its shipment aggregate
class CustomerRegistered(BaseEvent):
    customer_id = Identifier(required=True)
    name = String(required=True)


fulfillment_domain.register_external_event(
    CustomerRegistered, "Identity.CustomerRegistered.v1"
)


@fulfillment_domain.aggregate
class Shipment:
    recipient_id = Identifier(required=True)


@fulfillment_domain.event_handler(
    part_of=Shipment,
    stream_category="identity::customer",
)
class CustomerSyncHandler:
    @handle(CustomerRegistered)
    def on_registered(self, event: CustomerRegistered):
        recipient = Recipient(
            customer_id=event.customer_id,
            name=event.name,
        )
        current_domain.repository_for(Recipient).add(recipient)

Distributed domains: Subscribers as anti-corruption layers

When domains run as independent services with separate brokers (or when you consume events from external systems you don't control) use subscribers. Subscribers receive raw dict payloads and translate them into your domain's language, acting as an anti-corruption layer. Here the subscriber turns a customer payload into a CreateRecipient command, and a command handler stores the recipient:

from protean.fields import Text


@fulfillment_domain.command(part_of="Recipient")
class CreateRecipient:
    customer_id = Identifier(required=True)
    name = String(required=True)
    address = Text()


@fulfillment_domain.command_handler(part_of="Recipient")
class RecipientCommandHandler:
    @handle(CreateRecipient)
    def create_recipient(self, command: CreateRecipient) -> None:
        current_domain.repository_for(Recipient).add(
            Recipient(
                customer_id=command.customer_id,
                name=command.name,
                delivery_address=command.address,
            )
        )


@fulfillment_domain.subscriber(stream="identity_customer_events")
class CustomerEventSubscriber:
    """Anti-corruption layer: translates external customer events."""

    def __call__(self, payload: dict) -> None:
        event_type = payload.get("type")

        if event_type == "CustomerRegistered":
            current_domain.process(
                CreateRecipient(
                    customer_id=payload["customer_id"],
                    name=payload["full_name"],  # Field name translation
                    address=payload.get("shipping_address"),
                )
            )

Why this works for distributed domains:

  • Schema isolation: Your domain doesn't depend on the external domain's event classes. If the upstream renames a field, only the subscriber changes.
  • Translation at the boundary: External field names, types, and concepts are mapped to your domain's language before entering the domain.
  • Independent evolution: The upstream can deploy schema changes without coordinating with you. The subscriber absorbs the difference.

Key principles:

  • Subscribers receive dict, not typed events: Your domain doesn't import the external domain's classes
  • Translate at the boundary: Map external field names, types, and concepts to your domain's language
  • Everything downstream uses internal types: Only the subscriber knows about the external schema

See Consuming Events from Other Domains for the full pattern.

Fact events for state transfer

When an external consumer needs complete aggregate state (not granular deltas), enable fact events on the source aggregate:

from protean.core.aggregate import BaseAggregate
from protean.fields import String


@identity_domain.aggregate(fact_events=True)
class Customer(BaseAggregate):
    name = String(required=True)
    email = String(required=True)
    segment = String()

Fact events publish a full snapshot with every change, making downstream consumers simpler. They replace their local copy wholesale instead of applying incremental updates.

See Fact Events as Integration Contracts.

Correlation across contexts

The same real-world entity (e.g. a customer) often appears in multiple bounded contexts with different names and shapes. Link them using a shared identifier:

from protean.core.aggregate import BaseAggregate
from protean.fields import Auto


# Identity context: the authority for customer data
@identity_domain.aggregate
class Customer(BaseAggregate):
    customer_id = Auto(identifier=True)
    name = String(required=True)
    email = String(required=True)


# Fulfillment context: local representation with only relevant fields
@fulfillment_domain.aggregate
class Recipient(BaseAggregate):
    recipient_id = Auto(identifier=True)
    customer_id = Identifier(required=True)  # Correlation ID
    name = String(required=True)
    delivery_address = Text()

The customer_id in Recipient is a correlation ID. It links back to the authoritative Customer in the identity context without creating a code dependency.

See Connecting Concepts Across Bounded Contexts.

Running multiple domain servers

Each domain runs its own server process for async event/command processing:

# Terminal 1
protean server --domain my_app.identity

# Terminal 2
protean server --domain my_app.catalogue

# Terminal 3
protean server --domain my_app.fulfillment

Monitoring across domains

The observatory supports monitoring multiple domains simultaneously:

protean observatory \
    --domain my_app.identity \
    --domain my_app.catalogue \
    --domain my_app.fulfillment

Testing multi-domain applications

Separate fixtures per domain

Create independent test fixtures for each domain:

import pytest

from protean.integrations.pytest import DomainFixture

# Also import identity_domain and fulfillment_domain
# from my_app.identity and my_app.fulfillment.


@pytest.fixture(scope="session")
def identity_fixture():
    identity_domain.config["command_processing"] = "sync"
    identity_domain.config["event_processing"] = "sync"
    fixture = DomainFixture(identity_domain)
    fixture.setup()
    yield fixture
    fixture.teardown()


@pytest.fixture(scope="session")
def fulfillment_fixture():
    fulfillment_domain.config["command_processing"] = "sync"
    fulfillment_domain.config["event_processing"] = "sync"
    fixture = DomainFixture(fulfillment_domain)
    fixture.setup()
    yield fixture
    fixture.teardown()

Testing cross-domain flows

In tests the domains run in sync mode, and a domain in sync mode runs only its own handlers. An event raised in the identity domain never reaches a fulfillment handler, so a test cannot follow a flow across the boundary. Test each side of the boundary instead. On the receiving side, hand the event the other domain publishes to your handler, and check the effect in your domain:

def test_customer_registration_creates_recipient(fulfillment_fixture):
    with fulfillment_fixture.domain_context():
        # The event the identity domain publishes, as fulfillment defines it
        event = CustomerRegistered(customer_id="cust-1", name="Alice")
        CustomerSyncHandler().on_registered(event)

        repo = fulfillment_domain.repository_for(Recipient)
        recipients = repo.query.filter(name="Alice").all()
        assert recipients.total == 1
        assert recipients.first.customer_id == "cust-1"

To run a flow end to end across domains, run each domain's engine against a shared event store, as in production.

Testing FastAPI endpoints across domains

from fastapi.testclient import TestClient

# Also import app from my_app.api.app.


@pytest.fixture
def client():
    return TestClient(app)


def test_customer_endpoint_uses_identity_domain(client):
    response = client.post("/customers", json={"name": "Alice"})
    assert response.status_code == 201
    assert response.json() == {"domain": "Identity"}


def test_product_endpoint_uses_catalogue_domain(client):
    response = client.get("/products")
    assert response.status_code == 200
    assert response.json() == {"domain": "Catalogue"}

Design guidelines

  1. Start with one domain: Split only when you observe genuine language divergence or team boundaries. Premature splitting adds complexity without benefit.

  2. Each domain owns its data: Domains should not share databases. If two domains need the same data, one is the authority and the other holds a local copy synchronized through events.

  3. Communicate through events, not imports: Never import an aggregate from another domain's package. Use subscribers or cross-stream event handlers to react to changes.

  4. Translate at boundaries: Use the anti-corruption layer pattern (subscribers) to translate external concepts into your domain's language.

  5. Deploy independently when possible: Each domain should be deployable on its own schedule. Shared deployment couples teams and slows everyone down.