Skip to content

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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@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
    protocol_version: str
    method: str
    params: Mapping[str, Any] | None = None
    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 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
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

    @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
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.

CallNext module-attribute 

CallNext = Callable[
    ["ServerRequestContext[Any, Any]"],
    Awaitable[HandlerResult],
]

Invokes the rest of the chain. Pass the ctx through; rewrite method or params with dataclasses.replace(ctx, ...) to alter what the handler sees.

ServerMiddleware

Bases: Protocol[_MwLifespanT]

Context-tier middleware: (ctx, 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(ctx). 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.

The method and the raw inbound params are ctx.method and ctx.params (no model validation has happened yet). To rewrite either before the handler runs, pass an adjusted context: await call_next(replace(ctx, params=...)). ctx.request_id is None distinguishes a notification from a request. For notifications call_next(ctx) returns None (a dropped or unhandled notification also returns None) and the middleware's own return value is discarded.

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. initialize is observed but not rewritable: the post-chain handshake commit reads the wire params, so to veto the handshake raise before call_next().

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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class ServerMiddleware(Protocol[_MwLifespanT]):
    """Context-tier middleware: `(ctx, 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(ctx)`.
    `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`.

    The method and the raw inbound params are `ctx.method` and `ctx.params` (no
    model validation has happened yet). To rewrite either before the handler
    runs, pass an adjusted context: `await call_next(replace(ctx, params=...))`.
    `ctx.request_id is None` distinguishes a notification from a request. For
    notifications `call_next(ctx)` returns `None` (a dropped or unhandled
    notification also returns `None`) and the middleware's own return value is
    discarded.

    !!! 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.
        `initialize` is observed but not rewritable: the post-chain handshake
        commit reads the wire params, so to veto the handshake raise *before*
        `call_next()`.

    `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],
        call_next: CallNext,
    ) -> HandlerResult: ...