Middleware

Middleware is configured when starting the API layer. It runs for every event passing through the API boundary.

Configuring middleware

Middleware is provided when starting the API:

edri.start_api(
    middlewares=[AuthenticationMiddleware()],
)

• Middleware order is respected as provided.

Where middleware sits in the pipeline

Middleware participates in both the request path and the response path.

Request path

1. Incoming request/message is received (HTTP/WS/HTML)

2. An Event instance is constructed + populated (input extraction, conversions, validation/inject)

3. Middleware process_request runs

4. If middleware allows, the event proceeds to routing/handling

process_request is the last step before routing/handling.

EdriRegister and EdriUnregister also run process_request, even though the Broker answers them itself instead of routing them. Denying them uses the same convention as any other request: mark the response and continue.

Response path

After the router/handler produces an outgoing event/response for the client, the API layer applies middleware again:

1. Router/handler returns the event back to the API layer

2. Middleware process_response runs

3. API sends the final output to the client

Depending on the transport and implementation details, an event may traverse the router boundary more than once (e.g., request → router, response → router/API). The key guarantee is:

• process_request runs before routing/handling

• process_response runs when returning from routing/handling before sending to the client

Middleware and API types (REST / WS / HTML)

EDRI aims to keep middleware as generic as possible. Middleware should generally be written to work the same way regardless of whether an event arrived via REST, WS, or HTML.

In other words: middleware is transport-agnostic by design.

When you really need transport-specific behavior

Sometimes you may still want different behavior per transport (e.g., allow HTML to redirect but require REST to return JSON, or treat WS handshake differently).

EDRI intentionally does not expose API type through a public middleware context.

If you must branch by API type, the framework keeps the information on the event’s private API metadata:

def process_request(self, event: Event, info: EventInfo) -> MiddlewareControl:
    api_type = event._api.type

    if api_type == ApiType.WS:
        # WS-specific checks
        return MiddlewareControl.continue_processing()

    if api_type == ApiType.REST:
        # REST-specific checks
        return MiddlewareControl.continue_processing()

    if api_type == ApiType.HTML:
        # HTML-specific checks
        return MiddlewareControl.continue_processing()

    return MiddlewareControl.continue_processing()

Important notes

• event._api.type is private on purpose.

• In middleware you can assume _api is already populated, so event._api.type access is safe.

• Prefer transport-agnostic middleware whenever possible.

• If you branch on event._api.type, treat it as an advanced/internal dependency and keep usage centralized.

Transport-agnostic middleware

If you don’t explicitly branch by transport, middleware logic should be written as transport-agnostic. This is often desirable for shared concerns like authentication, rate-limiting, logging, and input hardening.

Middleware execution order

Middleware is executed in the following order:

• Request path: in the order provided to start_api

  • middlewares=[A(), B(), C()] → A.process_request → B.process_request → C.process_request

• Response path: reverse order (onion model)

  • middlewares=[A(), B(), C()] → C.process_response → B.process_response → A.process_response

Example: Authentication middleware via event inheritance

A common pattern is to use event inheritance to declare authentication requirements, and have middleware enforce them.

• BearerEvent → requires a Bearer token

• SessionEvent → requires a session cookie / session object

• AuthenticatedEvent → accepts either method

Event marker classes

from dataclasses import dataclass, field


@dataclass
class SessionEvent(Event):
    """Requires a valid session (e.g. JWE token in a cookie)."""

    cookie_session: str | None = field(default=Cookie("session"), kw_only=True)


@dataclass
class BearerEvent(Event):
    """Requires a valid Bearer token."""

    bearer_token: list[str] | None = field(default=Header("Authorization"), kw_only=True)


@dataclass
class AuthenticatedEvent(BearerEvent, SessionEvent):
    """Accepts either Bearer or Session authentication."""

    pass

Middleware implementation

from edri.api.middleware import Middleware, MiddlewareControl
from edri.dataclass.directive.base import UnauthorizedResponseDirective
from edri.dataclass.response import ResponseStatus


class AuthenticationMiddleware(Middleware):
    def process_request(self, event: Event, info: "EventInfo") -> MiddlewareControl:
        if isinstance(event, AuthenticatedEvent):
            session_ok, session_directive = self.check_session_event(event)
            bearer_ok, bearer_directive = self.check_bearer_event(event)

            if any((session_ok, bearer_ok)):
                return MiddlewareControl.continue_processing()

            event.response.set_status(ResponseStatus.FAILED)
            if session_directive is not None:
                event.response.add_directive(session_directive)
            if bearer_directive is not None:
                event.response.add_directive(bearer_directive)
            return MiddlewareControl.continue_processing()

        if isinstance(event, BearerEvent):
            ok, directive = self.check_bearer_event(event)
            if not ok:
                event.response.set_status(ResponseStatus.FAILED)
                if directive is not None:
                    event.response.add_directive(directive)
            return MiddlewareControl.continue_processing()

        if isinstance(event, SessionEvent):
            ok, directive = self.check_session_event(event)
            if not ok:
                event.response.set_status(ResponseStatus.FAILED)
                if directive is not None:
                    event.response.add_directive(directive)
            return MiddlewareControl.continue_processing()

        return MiddlewareControl.continue_processing()

    # --- helper methods (example signatures) ---

    def check_session_event(self, event: SessionEvent) -> tuple[bool, ResponseDirective | None]:
        """Return (ok, directive). If ok is False, directive should explain why."""
        ...

    def check_bearer_event(self, event: BearerEvent) -> tuple[bool, ResponseDirective | None]:
        """Return (ok, directive). If ok is False, directive should explain why."""
        ...

    # --- example helper method (bearer token check) ---

    def check_bearer_event(self, event: BearerEvent) -> tuple[bool, ResponseDirective | None]:
        if event.bearer_token is None or len(event.bearer_token) > 1:
            return False, UnauthorizedResponseDirective("Only one Authorization header is allowed")

        header = event.bearer_token[0]
        if header[:6].lower() != "bearer":
            return False, UnauthorizedResponseDirective("Authorization is supported only via a Bearer token")

        if header[7:].rstrip() != self.token:
            return False, UnauthorizedResponseDirective("Invalid bearer token")

        return True, None

process_response timing

process_response is called when the router/handler returns an event back to the API layer for delivery to the client. This allows middleware to:

• add/adjust directives (headers, cookies, redirects, status overrides)

• modify serialization-relevant fields

• perform post-processing (logging, audit)

Registering custom events (register_events)

Normally middleware only sees events that some Manager already handles. To let middleware intercept events that no Manager handles — typically a custom handshake or control event — override register_events:

class WsAuthMiddleware(Middleware):
    @staticmethod
    def register_events() -> tuple[list[Type[Event]], list[Type[Event]]]:
        # (request events, response events)
        return [WsLogin], []

    def process_request(self, event: Event, info: "EventInfo | None") -> MiddlewareControl:
        if isinstance(event, WsLogin):
            session = authenticate(event)
            # turn the handshake into a real flow
            return MiddlewareControl.replace_event(LoginResult(session=session))
        return MiddlewareControl.continue_processing()

It returns a (requests, responses) tuple of event classes. At startup the Broker subscribes itself to those types on the Router, so an incoming event of that type is routed to the Broker even though no Manager subscribes to it. The Broker then runs only this middleware on the event and acts on the returned MiddlewareControl.

This is what makes WebSocket authentication possible: a client can send a login event before any Manager would accept it, and the middleware can replace_event or redirect_event it into the authenticated flow (or stop_processing to reject it).

It also pairs with nested events: the registered event can carry the client's original event embedded inside it, so the middleware unwraps the inner event once authentication succeeds and forwards it on — see the Nested event model. This keeps the original request intact while the handshake event drives the auth step around it.

MiddlewareControl Actions

Action	Behavior
CONTINUE	Continue processing
STOP	Drop event immediately
REPLACE_EVENT	Replace current event
REDIRECT	Reroute event internally

STOP Behavior

When STOP is returned, the event is discarded and no response is sent. For REST/HTML requests this leaves the caller waiting until the API response timeout. To reject a request with an HTTP status, use REDIRECT with a completed response instead.

REDIRECT Behavior

REDIRECT changes the processing direction. From process_request, it sends a completed response through the response middleware and back to the client, without invoking a Manager. From process_response, it sends the supplied request to the Router. The direction is determined before the middleware runs, so setting the response on the original event is supported.

For example, to reject an unauthorized request immediately:

from edri.api.broker import EventInfo
from edri.api.middleware import Middleware, MiddlewareControl
from edri.dataclass.directive.base import UnauthorizedResponseDirective
from edri.dataclass.event import Event
from edri.dataclass.response import Response, ResponseStatus


class Reject(Middleware):
    def process_request(self, event: Event, info: EventInfo | None) -> MiddlewareControl:
        assert isinstance(event.response, Response)
        event.response.set_status(ResponseStatus.FAILED)
        event.response.add_directive(UnauthorizedResponseDirective("no"))
        return MiddlewareControl.redirect_event(event)

On a REST endpoint this returns HTTP 401 immediately. The endpoint must declare a response type. Response middleware runs in reverse order as usual.