Quick start

This tutorial builds one complete request flow: PlaceOrder goes into a mediator, PlaceOrderHandler handles it, and an OrderReceipt comes back.

You need Python 3.12 or later and an installed copy of PyMediate.

Create the program

Save the following code as app.py:

import asyncio
from dataclasses import dataclass

from pymediate import Mediator, Request, RequestHandler, Services


@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}",
        )


async def main() -> None:
    services = Services(PlaceOrderHandler())
    mediator = Mediator(services)

    receipt = await mediator.send(
        PlaceOrder(customer_id=7, item="tea", quantity=2),
    )
    print(f"Order {receipt.order_id}: {receipt.summary}")


asyncio.run(main())

The fixed order ID keeps this first example focused on dispatch. A real handler can receive a database or another dependency through its constructor.

Run it

python app.py

The program prints:

Order 42: 2 × tea

Read the declarations

The three class declarations describe the complete type relationship:

1. OrderReceipt is an ordinary dataclass. PyMediate does not require a response base class.
2. PlaceOrder(Request[OrderReceipt]) declares that sending PlaceOrder returns an OrderReceipt.
3. PlaceOrderHandler(RequestHandler[PlaceOrder]) declares that the handler accepts PlaceOrder.

The introduction's type guide shows these relationships visually.

Follow the dispatch

These two lines create the mediator:

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

Services holds the handler instance and is itself the read-only ServiceProvider the mediator uses for resolution, so it is passed straight to Mediator.

Sending the request then follows four steps:

1. send() receives a PlaceOrder instance.
2. The mediator finds the handler type registered for the exact PlaceOrder class.
3. The service provider returns the PlaceOrderHandler instance.
4. The mediator awaits the handler and returns its OrderReceipt.

Because PlaceOrder inherits from Request[OrderReceipt], a type checker infers receipt as OrderReceipt. PyMediate separately validates the handler's parameter and return annotations when Python defines PlaceOrderHandler.

Use the synchronous API

The top-level package is asynchronous. For synchronous code, import Mediator, RequestHandler, and Services from pymediate.sync, define the handler with def, and call mediator.send() without await.

The async and sync guide contains complete asynchronous and synchronous programs.

Continue

• Read the core concepts for notifications, streaming, pipeline behaviors, and service providers.
• Add dependencies and test handlers directly with the handlers guide.
• Look up exact constraints in the RequestHandler reference.