refactor: remove mixins; inline typed requests into their owners

Connection now defines its typed send_request ladder directly,
ClientPeer is a plain class owning the peer request methods, and the
unwired Context drops that surface entirely. No method bodies were
duplicated; net deletion.

Author: maxisbey

src/mcp/server/_typed_request.py

@@ -16,20 +16,47 @@
 import logging
 from collections.abc import Mapping
 from contextlib import AsyncExitStack
-from typing import Any
+from typing import Any, TypeVar, overload
 
 import anyio
+from pydantic import BaseModel
 
-from mcp.server._typed_request import TypedServerRequestMixin
 from mcp.shared.dispatcher import CallOptions, Outbound
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.peer import Meta, dump_params
-from mcp.types import ClientCapabilities, InitializeRequestParams, LoggingLevel
+from mcp.types import (
+    ClientCapabilities,
+    CreateMessageRequest,
+    CreateMessageResult,
+    ElicitRequest,
+    ElicitResult,
+    EmptyResult,
+    InitializeRequestParams,
+    ListRootsRequest,
+    ListRootsResult,
+    LoggingLevel,
+    PingRequest,
+    Request,
+)
 
 __all__ = ["Connection"]
 
 logger = logging.getLogger(__name__)
 
+ResultT = TypeVar("ResultT", bound=BaseModel)
+
+# Result types for the spec's server-to-client request set, used by
+# `Connection.send_request` to infer the result type. If the spec's request
+# set grows substantially, consider declaring the result mapping on the
+# request types themselves (a `__mcp_result__` ClassVar read via a structural
+# protocol) so this table and the overload ladder don't need maintaining.
+_RESULT_FOR: dict[type[Request[Any, Any]], type[BaseModel]] = {
+    CreateMessageRequest: CreateMessageResult,
+    ElicitRequest: ElicitResult,
+    ListRootsRequest: ListRootsResult,
+    PingRequest: EmptyResult,
+}
+
 
 def _notification_params(payload: dict[str, Any] | None, meta: Meta | None) -> dict[str, Any] | None:
     if not meta:
@@ -39,7 +66,7 @@ def _notification_params(payload: dict[str, Any] | None, meta: Meta | None) -> d
     return out
 
 
-class Connection(TypedServerRequestMixin):
+class Connection:
     """Per-client connection state and standalone-stream `Outbound`.
 
     Constructed by `ServerRunner` once per connection. The peer-info fields
@@ -98,10 +125,10 @@ async def send_raw_request(
     ) -> dict[str, Any]:
         """Send a raw request on the standalone stream.
 
-        Low-level `Outbound` channel. Prefer the typed `send_request` (from
-        `TypedServerRequestMixin`) or the convenience methods below; use this
-        directly only for off-spec messages. `opts` carries per-call `timeout`
-        / `on_progress` / resumption hints; see `CallOptions`.
+        Low-level `Outbound` channel. Prefer the typed `send_request` or the
+        convenience methods below; use this directly only for off-spec
+        messages. `opts` carries per-call `timeout` / `on_progress` /
+        resumption hints; see `CallOptions`.
 
         Raises:
             MCPError: The peer responded with an error.
@@ -111,6 +138,42 @@ async def send_raw_request(
             raise NoBackChannelError(method)
         return await self._outbound.send_raw_request(method, params, opts)
 
+    @overload
+    async def send_request(
+        self, req: CreateMessageRequest, *, opts: CallOptions | None = None
+    ) -> CreateMessageResult: ...
+    @overload
+    async def send_request(self, req: ElicitRequest, *, opts: CallOptions | None = None) -> ElicitResult: ...
+    @overload
+    async def send_request(self, req: ListRootsRequest, *, opts: CallOptions | None = None) -> ListRootsResult: ...
+    @overload
+    async def send_request(self, req: PingRequest, *, opts: CallOptions | None = None) -> EmptyResult: ...
+    @overload
+    async def send_request(
+        self, req: Request[Any, Any], *, result_type: type[ResultT], opts: CallOptions | None = None
+    ) -> ResultT: ...
+    async def send_request(
+        self,
+        req: Request[Any, Any],
+        *,
+        result_type: type[BaseModel] | None = None,
+        opts: CallOptions | None = None,
+    ) -> BaseModel:
+        """Send a typed server-to-client request and return its typed result.
+
+        For spec request types the result type is inferred. For custom requests
+        pass `result_type=` explicitly.
+
+        Raises:
+            MCPError: The peer responded with an error.
+            NoBackChannelError: No back-channel for server-initiated requests.
+            pydantic.ValidationError: The peer's result does not match the expected result type.
+            KeyError: `result_type` omitted for a non-spec request type.
+        """
+        raw = await self.send_raw_request(req.method, dump_params(req.params), opts)
+        cls = result_type if result_type is not None else _RESULT_FOR[type(req)]
+        return cls.model_validate(raw, by_name=False)
+
     async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
         """Send a best-effort notification on the standalone stream.
 

src/mcp/server/connection.py

@@ -5,13 +5,12 @@
 from pydantic import BaseModel
 from typing_extensions import TypeVar
 
-from mcp.server._typed_request import TypedServerRequestMixin
 from mcp.server.connection import Connection
 from mcp.server.session import ServerSession
 from mcp.shared.context import BaseContext
 from mcp.shared.dispatcher import DispatchContext
 from mcp.shared.message import CloseSSEStreamCallback
-from mcp.shared.peer import ClientPeerMixin, Meta
+from mcp.shared.peer import Meta
 from mcp.shared.transport_context import TransportContext
 from mcp.types import LoggingLevel, RequestId, RequestParamsMeta
 
@@ -43,16 +42,14 @@ class ServerRequestContext(Generic[LifespanContextT, RequestT]):
 LifespanT_co = TypeVar("LifespanT_co", default=Any, covariant=True)
 
 
-class Context(BaseContext[TransportContext], ClientPeerMixin, TypedServerRequestMixin, Generic[LifespanT_co]):
+class Context(BaseContext[TransportContext], Generic[LifespanT_co]):
     """Server-side per-request context.
 
-    Composes `BaseContext` (forwards to `DispatchContext`, satisfies `Outbound`),
-    `ClientPeerMixin` (kwarg-style `sample`/`elicit_*`/`list_roots`/`ping`),
-    and `TypedServerRequestMixin` (typed `send_request(req) -> Result`). Adds
-    `lifespan` and `connection`.
+    Extends `BaseContext` (transport metadata, the raw back-channel, progress
+    reporting) with `lifespan`, `connection`, and request-scoped `log`.
 
-    Constructed by `ServerRunner` per inbound request and handed to the user's
-    handler.
+    Not currently constructed by `ServerRunner`, which hands handlers a
+    `ServerRequestContext` instead.
     """
 
     def __init__(

src/mcp/server/context.py

@@ -4,10 +4,9 @@
 back-channel (`send_raw_request`/`notify`), progress reporting, and the cancel
 event. Adds `meta` (the inbound request's `_meta` field).
 
-Satisfies `Outbound`, so `ClientPeerMixin` works on it (the server-side `Context`
-mixes that in directly). Shared between client and server: the server's
-`Context` extends this with `lifespan`/`connection`; `ClientContext` is just an
-alias.
+Satisfies `Outbound`, so `ClientPeer` can wrap it. Shared between client and
+server: the server's `Context` extends this with `lifespan`/`connection`;
+`ClientContext` is just an alias.
 """
 
 from collections.abc import Mapping

src/mcp/shared/context.py

@@ -84,9 +84,9 @@ class Outbound(Protocol):
     """Anything that can send requests and notifications to the peer.
 
     Both `Dispatcher` (top-level outbound) and `DispatchContext` (back-channel
-    during an inbound request) extend this. The MCP type layer (`ClientPeerMixin`,
-    `Connection`, `Context`) builds typed `send_request` / convenience methods
-    on top of this raw channel.
+    during an inbound request) extend this. The MCP type layer (`ClientPeer`,
+    `Connection`) builds typed `send_request` / convenience methods on top of
+    this raw channel.
     """
 
     async def send_raw_request(