Introduction

PyMediate is a Python 3.12+ library that routes typed requests to handlers. A request declares its response type, so Mediator.send() preserves that type for static type checkers and editors.

The core flow has three declarations:

from dataclasses import dataclass
from pymediate import Request, RequestHandler

@dataclass(frozen=True)
class OrderReceipt:
    order_id: int
    summary: str

@dataclass(frozen=True)
class PlaceOrder(Request[OrderReceipt]):
    customer_id: int
    item: str
    quantity: int

class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        return OrderReceipt(
            order_id=42,
            summary=f"{request.quantity} × {request.item}",
        )

How to read the types

The value inside each pair of square brackets states a relationship. The following guide highlights the code and its plain-language reading together.

class OrderReceipt:
    ...

class PlaceOrder(Request[OrderReceipt]):
    ...

class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    ...

Request[OrderReceipt] records the response type on PlaceOrder. RequestHandler[PlaceOrder] records which request the handler accepts. PyMediate validates the handler's annotated parameter and return types when Python defines the class.

Sending the request

Register a handler instance, create a mediator, and send the request:

from pymediate import Mediator, Services

services = Services(PlaceOrderHandler())
mediator = Mediator(services)

receipt = await mediator.send(
    PlaceOrder(customer_id=7, item="tea", quantity=2),
)

print(receipt.order_id)
# 42

At dispatch, the mediator uses the request's exact class to find its handler type, asks the service provider for the registered handler instance, and calls it. The type of receipt is inferred as OrderReceipt from PlaceOrder(Request[OrderReceipt]).

The quick start turns this example into a complete runnable program and explains each step.

The rest of the API

Start with send(). The other dispatch forms use the same typed-message model for different result shapes:

PipelineBehavior can wrap send() with shared processing such as logging or validation. Services and the ServiceProvider protocol supply handler and behavior instances to the mediator. The core concepts page introduces these parts after the request flow.

The top-level pymediate package is asynchronous. pymediate.sync provides the corresponding blocking API for synchronous applications.

Scope

PyMediate provides in-process dispatch, type propagation at call sites, and class-definition-time validation of handler annotations. It does not provide a task queue, choose how your application stores data, or require CQRS or hexagonal architecture.

Direct calls are often clearer when callers can depend on their collaborators without repeated wiring. A small hand-written dispatcher can also be enough. Using a mediator to reduce change coupling develops the architectural case for a mediator and describes the trade-offs.

Where to begin