context

ServerRequestContext dataclass

Bases: Generic[LifespanContextT, RequestT]

Per-request context handed to lowlevel request and notification handlers.

Built by ServerRunner._make_context for each inbound message. Carries the connection-scoped ServerSession (server-to-client requests and notifications), per-request metadata, and any per-message data the transport attached (the HTTP request, SSE stream-close callbacks).

Source code in src/mcp/server/context.py

@dataclass(kw_only=True)
class ServerRequestContext(Generic[LifespanContextT, RequestT]):
    """Per-request context handed to lowlevel request and notification handlers.

    Built by `ServerRunner._make_context` for each inbound message. Carries the
    connection-scoped `ServerSession` (server-to-client requests and
    notifications), per-request metadata, and any per-message data the
    transport attached (the HTTP request, SSE stream-close callbacks).
    """

    session: ServerSession
    lifespan_context: LifespanContextT
    request_id: RequestId | None = None
    meta: RequestParamsMeta | None = None
    request: RequestT | None = None
    close_sse_stream: CloseSSEStreamCallback | None = None
    close_standalone_sse_stream: CloseSSEStreamCallback | None = None

Context

Bases: BaseContext[TransportContext], Generic[LifespanT_co]

Server-side per-request context.

Extends BaseContext (transport metadata, the raw back-channel, progress reporting) with lifespan, connection, and request-scoped log.

Not currently constructed by ServerRunner, which hands handlers a ServerRequestContext instead.

Source code in src/mcp/server/context.py

class Context(BaseContext[TransportContext], Generic[LifespanT_co]):
    """Server-side per-request context.

    Extends `BaseContext` (transport metadata, the raw back-channel, progress
    reporting) with `lifespan`, `connection`, and request-scoped `log`.

    Not currently constructed by `ServerRunner`, which hands handlers a
    `ServerRequestContext` instead.
    """

    def __init__(
        self,
        dctx: DispatchContext[TransportContext],
        *,
        lifespan: LifespanT_co,
        connection: Connection,
        meta: RequestParamsMeta | None = None,
    ) -> None:
        super().__init__(dctx, meta=meta)
        self._lifespan = lifespan
        self._connection = connection

    @property
    def lifespan(self) -> LifespanT_co:
        """The server-wide lifespan output (what `Server(..., lifespan=...)` yielded)."""
        return self._lifespan

    @property
    def connection(self) -> Connection:
        """The per-client `Connection` for this request's connection."""
        return self._connection

    @property
    def session_id(self) -> str | None:
        """The transport's session id for this connection, when one exists.

        Convenience for `ctx.connection.session_id`. `None` on stdio and
        stateless HTTP.
        """
        return self._connection.session_id

    @property
    def headers(self) -> Mapping[str, str] | None:
        """Request headers carried by this message, when the transport has them.

        Convenience for `ctx.transport.headers`. `None` on stdio.
        """
        return self.transport.headers

    async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *, meta: Meta | None = None) -> None:
        """Send a request-scoped `notifications/message` log entry.

        Uses this request's back-channel (so the entry rides the request's SSE
        stream in streamable HTTP), not the standalone stream - use
        `ctx.connection.log(...)` for that.
        """
        params: dict[str, Any] = {"level": level, "data": data}
        if logger is not None:
            params["logger"] = logger
        if meta:
            params["_meta"] = meta
        await self.notify("notifications/message", params)

lifespan property

lifespan: LifespanT_co

The server-wide lifespan output (what Server(..., lifespan=...) yielded).

connection property

connection: Connection

The per-client Connection for this request's connection.

session_id property

session_id: str | None

The transport's session id for this connection, when one exists.

Convenience for ctx.connection.session_id. None on stdio and stateless HTTP.

headers property

headers: Mapping[str, str] | None

Request headers carried by this message, when the transport has them.

Convenience for ctx.transport.headers. None on stdio.

log async

log(
    level: LoggingLevel,
    data: Any,
    logger: str | None = None,
    *,
    meta: Meta | None = None
) -> None

Send a request-scoped notifications/message log entry.

Uses this request's back-channel (so the entry rides the request's SSE stream in streamable HTTP), not the standalone stream - use ctx.connection.log(...) for that.

Source code in src/mcp/server/context.py

async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *, meta: Meta | None = None) -> None:
    """Send a request-scoped `notifications/message` log entry.

    Uses this request's back-channel (so the entry rides the request's SSE
    stream in streamable HTTP), not the standalone stream - use
    `ctx.connection.log(...)` for that.
    """
    params: dict[str, Any] = {"level": level, "data": data}
    if logger is not None:
        params["logger"] = logger
    if meta:
        params["_meta"] = meta
    await self.notify("notifications/message", params)

HandlerResult module-attribute

HandlerResult = BaseModel | dict[str, Any] | None

What a request handler (or middleware) may return. ServerRunner serializes all three to a result dict.

ServerMiddleware

Bases: Protocol[_MwLifespanT]

Context-tier middleware: (ctx, method, params, call_next) -> result.

Runs at the top of ServerRunner._on_request / _on_notify after ctx is built but before any validation, lookup, or handshake. Wraps every inbound request and notification: initialize, the pre-init gate, METHOD_NOT_FOUND, params validation, the handler call, and notifications/initialized all run inside call_next(). notifications/cancelled is observed too; the dispatcher applies the cancellation itself, then forwards the notification. A request-side failure reaches the middleware as a raised MCPError (or ValidationError for malformed params) so observation/logging middleware can record it. Listed outermost-first on Server.middleware.

ctx.request_id is None distinguishes a notification from a request. For notifications call_next() returns None (a dropped or unhandled notification also returns None) and the middleware's own return value is discarded.

params is the raw inbound mapping (no model validation has happened yet). For typed inspection, validate against the model the middleware expects.

Warning: initialize is handled inline - the dispatcher does not read further inbound messages until the middleware chain returns. Awaiting a server-to-client request (ctx.session.send_request, send_ping, ...) while handling initialize therefore deadlocks the connection: the response can never be dequeued. Send-and-forget notifications are safe.

Server[L].middleware holds ServerMiddleware[L], so an app-specific middleware sees ctx.lifespan_context: L. While the context is the mutable ServerRequestContext dataclass it is invariant in L, so a reusable middleware should be typed ServerMiddleware[Any] to register on any Server[L].

Source code in src/mcp/server/context.py

class ServerMiddleware(Protocol[_MwLifespanT]):
    """Context-tier middleware: `(ctx, method, params, call_next) -> result`.

    Runs at the top of `ServerRunner._on_request` / `_on_notify` after `ctx`
    is built but before any validation, lookup, or handshake. Wraps every
    inbound request and notification: `initialize`, the pre-init gate,
    `METHOD_NOT_FOUND`, params validation, the handler call, and
    `notifications/initialized` all run inside `call_next()`.
    `notifications/cancelled` is observed too; the dispatcher applies the
    cancellation itself, then forwards the notification. A request-side
    failure reaches the middleware as a raised `MCPError` (or
    `ValidationError` for malformed params) so observation/logging middleware
    can record it. Listed outermost-first on `Server.middleware`.

    `ctx.request_id is None` distinguishes a notification from a request. For
    notifications `call_next()` returns `None` (a dropped or unhandled
    notification also returns `None`) and the middleware's own return value is
    discarded.

    `params` is the raw inbound mapping (no model validation has happened
    yet). For typed inspection, validate against the model the middleware
    expects.

    Warning: `initialize` is handled inline - the dispatcher does not read
    further inbound messages until the middleware chain returns. Awaiting a
    server-to-client request (`ctx.session.send_request`, `send_ping`, ...)
    while handling `initialize` therefore deadlocks the connection: the
    response can never be dequeued. Send-and-forget notifications are safe.

    `Server[L].middleware` holds `ServerMiddleware[L]`, so an app-specific
    middleware sees `ctx.lifespan_context: L`. While the context is the
    mutable `ServerRequestContext` dataclass it is invariant in `L`, so a
    reusable middleware should be typed `ServerMiddleware[Any]` to register on
    any `Server[L]`.
    """

    # TODO(maxisbey): once `_make_context` returns the (covariant) `Context[L]`
    # again, restore `_MwLifespanT` to `contravariant=True` and retype `ctx`
    # below to `Context[_MwLifespanT]` so reusable middleware can be
    # `ServerMiddleware[object]` instead of `ServerMiddleware[Any]`.

    async def __call__(
        self,
        ctx: ServerRequestContext[_MwLifespanT, Any],
        method: str,
        params: Mapping[str, Any] | None,
        call_next: CallNext,
    ) -> HandlerResult: ...