Middleware: Token authentication end-to-end

A complete, self-contained example demonstrating how to protect REST endpoints with Bearer-token authentication using middleware and event inheritance.

The example shows:

• defining authenticated event base classes

• implementing an authentication middleware

• protecting individual endpoints

• returning proper 401 responses for unauthenticated requests

---

Event base classes

# events/auth.py
from dataclasses import dataclass, field

from edri.api.dataclass.api_event import Header
from edri.dataclass.event import Event


@dataclass
class BearerEvent(Event):
    """Requires a valid Bearer token in the Authorization header."""
    bearer_token: list[str] | None = field(
        default=Header("Authorization"),
        kw_only=True,
    )

Any event class that inherits from BearerEvent will automatically have the bearer_token field populated from the HTTP Authorization header by the API layer.

---

Protected events

# events/items.py
from http import HTTPMethod
from edri.api.dataclass.api_event import api, ApiType
from edri.dataclass.event import event
from edri.dataclass.response import Response, response
from events.auth import BearerEvent


@response
class ItemResponse(Response):
    id: int
    name: str


@api(url="/items/{item_id}", method=HTTPMethod.GET, exclude={ApiType.HTML, ApiType.WS})
class ItemGet(BearerEvent):
    item_id: int
    response: ItemResponse


@api(url="/public/info", method=HTTPMethod.GET, exclude={ApiType.HTML, ApiType.WS})
class PublicInfo(Event):
    response: ItemResponse   # does NOT inherit BearerEvent — public endpoint

---

Middleware

# middleware/auth.py
from edri.api.middleware import Middleware, MiddlewareControl
from edri.dataclass.directive.base import UnauthorizedResponseDirective
from edri.dataclass.response import ResponseStatus
from events.auth import BearerEvent

VALID_TOKEN = "supersecret"


class BearerAuthMiddleware(Middleware):

    def process_request(self, event, info) -> MiddlewareControl:
        if not isinstance(event, BearerEvent):
            return MiddlewareControl.continue_processing()

        token = self._extract_token(event.bearer_token)
        if token != VALID_TOKEN:
            event.response.set_status(ResponseStatus.FAILED)
            event.response.add_directive(
                UnauthorizedResponseDirective("Invalid or missing Bearer token")
            )
            # CONTINUE lets the Broker send the 401 response back to the client.
            # Use STOP only when you want to drop the connection silently.
            return MiddlewareControl.continue_processing()

        return MiddlewareControl.continue_processing()

    @staticmethod
    def _extract_token(header_values: list[str] | None) -> str | None:
        if not header_values or len(header_values) != 1:
            return None
        value = header_values[0]
        if not value.lower().startswith("bearer "):
            return None
        return value[7:].strip()

---

Manager

# managers/items.py
from edri.abstract import ManagerBase, request
from edri.dataclass.directive.http import NotFoundResponseDirective
from edri.dataclass.response import ResponseStatus

from events.items import ItemGet, PublicInfo

_ITEMS = {
    1: {"id": 1, "name": "widget"},
    2: {"id": 2, "name": "gadget"},
}


class ItemsManager(ManagerBase):

    @request
    def get_item(self, event: ItemGet) -> None:
        item = _ITEMS.get(event.item_id)
        if item is None:
            event.response.set_status(ResponseStatus.FAILED)
            event.response.add_directive(NotFoundResponseDirective())
            return
        event.response.id = item["id"]
        event.response.name = item["name"]

    @request
    def public_info(self, event: PublicInfo) -> None:
        event.response.id = 0
        event.response.name = "public"

---

Entry point

# main.py
from edri import EDRI
from managers.items import ItemsManager
from middleware.auth import BearerAuthMiddleware

if __name__ == "__main__":
    edri = EDRI()
    edri.add_component(ItemsManager(edri.router_queue))
    edri.start_api(middlewares=[BearerAuthMiddleware()])
    edri.run()

---

Trying it out

# Should succeed — valid token
curl -H "Accept: application/json" \
     -H "Authorization: Bearer supersecret" \
     http://localhost:8080/items/1

# Should return 401 — wrong token
curl -H "Accept: application/json" \
     -H "Authorization: Bearer wrongtoken" \
     http://localhost:8080/items/1

# Should return 401 — no token at all
curl -H "Accept: application/json" \
     http://localhost:8080/items/1

# Should succeed — public endpoint, no token required
curl -H "Accept: application/json" \
     http://localhost:8080/public/info

Expected 401 response. Because the UnauthorizedResponseDirective maps to HTTP 401 (a client error), the Listener renders the standard error envelope (via response_error), not the normal data/status body:

{
  "status_code": 401,
  "status": "UNAUTHORIZED",
  "description": "No permission -- see authorization schemes",
  "reasons": [
    {
      "message": "Invalid or missing Bearer token",
      "status_code": 401
    }
  ]
}

The UnauthorizedResponseDirective sets the HTTP status to 401; the message you pass it appears under reasons. (In production, EDRI_API_EXPOSE_ERROR_DETAILS=false keeps only the first reason.) The EDRI internal ResponseStatus.FAILED set by the middleware is separate from the HTTP status code — see Response.

---

Notes

• BearerAuthMiddleware is transport-agnostic — it will also protect WebSocket events that inherit from BearerEvent, if any.

• Extend BearerEvent to add more auth requirements (session cookies, API key, etc.) rather than adding logic inside the middleware.

• In production, load VALID_TOKEN from an environment variable or a secrets manager, not a hardcoded constant.

• The process_response method is not implemented here — middleware can also log or modify outgoing responses; see Middleware for patterns.