Support vendor request types in ClientSession.send_request

Request gains a name_param ClassVar naming the wire-params key to mirror
into the Mcp-Name header; send_request emits it whenever the stamp has not
already set the header, so core NAME_BEARING_METHODS rows win by ordering
and a missing value fails loud instead of silently omitting the header.
send_request's typing widens to accept any Request[Any, Any] (runtime was
always duck-typed), retiring the cast ceremony in the extension and
custom-method stories. dispatch_input_request and validate_tool_result
are promoted to public ClientSession surface.

Author: maxisbey

docs/advanced/extensions.md

@@ -119,8 +119,8 @@ The same file's `main()` is the whole client story, both halves of it:
   which: `require_client_extension(ctx, ...)` and
   `ctx.session.check_client_capability(...)` read the right source on both paths.
 * Vendor methods drop one layer to `client.session.send_request(...)`; `Client`
-  only grows first-class methods for spec verbs. The `cast` is there because
-  `send_request` is typed against the spec's closed request union.
+  only grows first-class methods for spec verbs. `send_request` accepts any
+  `Request` subclass, so the vendor request passes as-is.
 
 ### Intercepting `tools/call`
 

docs_src/extensions/tutorial004.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from typing import Any, Literal, cast
+from typing import Any, Literal
 
 import mcp_types as types
 from pydantic import Field
@@ -53,6 +53,6 @@ def methods(self) -> Sequence[MethodBinding]:
 async def main() -> None:
     async with Client(mcp, extensions={EXTENSION_ID: {}}) as client:
         request = SearchRequest(params=SearchParams(query="mcp", limit=3))
-        result = await client.session.send_request(cast("types.ClientRequest", request), SearchResult)
+        result = await client.session.send_request(request, SearchResult)
         print(result.items)
         # ['mcp-0', 'mcp-1', 'mcp-2']

examples/stories/custom_methods/README.md

@@ -28,10 +28,9 @@ uv run python -m stories.custom_methods.client --http
   method string is the wire `method`; use a vendor prefix so it can never
   collide with a future spec method.
 - `client.py` `client.session.send_request(...)` — `Client` only exposes spec
-  verbs, so vendor methods go through the underlying `ClientSession`. The
-  `cast("types.ClientRequest", ...)` is needed because `send_request`'s
-  `request` parameter is currently typed as the closed spec union; widening it
-  (or adding `Client.send_request`) is tracked for beta.
+  verbs, so vendor methods go through the underlying `ClientSession`.
+  `send_request` accepts any `types.Request` subclass, so the vendor request
+  passes as-is, no cast.
 
 ## Caveats
 

examples/stories/custom_methods/client.py

@@ -1,6 +1,6 @@
 """Send a vendor-prefixed request via the `client.session` escape hatch."""
 
-from typing import Literal, cast
+from typing import Literal
 
 import mcp_types as types
 
@@ -26,12 +26,11 @@ async def main(target: Target, *, mode: str = "auto") -> None:
     async with Client(target, mode=mode) as client:
         # `Client` only exposes spec-defined verbs, so vendor methods have to drop one
         # layer to `client.session` today — there is no `Client`-level API for them
-        # yet, and whether `.session` stays public is undecided. `send_request` is
-        # typed against the closed `ClientRequest` union, hence the cast; at runtime
-        # the body only calls `.model_dump()` and the unknown method skips the
-        # per-spec result-validation registry.
+        # yet, and whether `.session` stays public is undecided. `send_request`
+        # accepts any `Request` subclass; the unknown method skips the per-spec
+        # result-validation registry.
         request = SearchRequest(params=SearchParams(query="mcp", limit=3))
-        result = await client.session.send_request(cast("types.ClientRequest", request), SearchResult)
+        result = await client.session.send_request(request, SearchResult)
         assert result.items == ["mcp-0", "mcp-1", "mcp-2"], result
 
 

examples/stories/extensions/client.py

@@ -1,6 +1,6 @@
 """Discover an extension's capability entry, call its tool, then send its vendor method."""
 
-from typing import Literal, cast
+from typing import Literal
 
 import mcp_types as types
 from mcp_types import TextContent
@@ -43,10 +43,9 @@ async def main(target: Target, *, mode: str = "auto") -> None:
         assert isinstance(result.content[0], TextContent)
         assert result.content[0].text == "mcp-suggestion", result.content[0].text
 
-        # Vendor methods drop one layer to `client.session` (see custom_methods/);
-        # the cast is needed because `send_request` is typed against the spec union.
+        # Vendor methods drop one layer to `client.session` (see custom_methods/).
         request = SearchRequest(params=SearchParams(query="mcp", limit=3))
-        found = await client.session.send_request(cast("types.ClientRequest", request), SearchResult)
+        found = await client.session.send_request(request, SearchResult)
         assert found.items == ["mcp-0", "mcp-1", "mcp-2"], found
 
 

src/mcp-types/mcp_types/_types.py

@@ -8,7 +8,7 @@
 
 from __future__ import annotations
 
-from typing import Annotated, Any, Final, Generic, Literal, TypeAlias, TypeVar
+from typing import Annotated, Any, ClassVar, Final, Generic, Literal, TypeAlias, TypeVar
 
 from pydantic import (
     BaseModel,
@@ -128,6 +128,12 @@ class Request(MCPModel, Generic[RequestParamsT, MethodT]):
     method: MethodT
     params: RequestParamsT
 
+    name_param: ClassVar[str | None] = None
+    """Wire-params key mirrored into the `Mcp-Name` header on sends (SEP-2243
+    family; SEP-2663 requires it for tasks/*). The request type declares; the
+    host emits. Subclasses override by bare assignment (`name_param = "taskId"`)
+    — re-annotating as `ClassVar[str]` trips pyright's ClassVar invariance."""
+
 
 class PaginatedRequest(Request[PaginatedRequestParams | None, MethodT], Generic[MethodT]):
     """Base class for paginated requests, matching the schema's PaginatedRequest interface."""

src/mcp/client/client.py

@@ -715,7 +715,7 @@ async def _drive_input_required(
 
         async def dispatch(key: str, req: InputRequest) -> InputResponse | ErrorData:
             ctx = ClientRequestContext(session=session, request_id=key, meta=req.params.meta if req.params else None)
-            return await session._dispatch_input_request(ctx, req)  # pyright: ignore[reportPrivateUsage]
+            return await session.dispatch_input_request(ctx, req)
 
         return await run_input_required_driver(
             first, dispatch=dispatch, retry=retry, max_rounds=self.input_required_max_rounds

src/mcp/client/session.py

@@ -304,7 +304,7 @@ async def __aexit__(
 
     async def send_request(
         self,
-        request: types.ClientRequest,
+        request: types.ClientRequest | types.Request[Any, Any],
         result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT],
         request_read_timeout_seconds: float | None = None,
         metadata: ClientMessageMetadata | None = None,
@@ -318,11 +318,22 @@ async def send_request(
         Raises:
             MCPError: Error response, read timeout, or connection closed.
             RuntimeError: Called before entering the context manager.
+            ValueError: The request type declares `name_param` but the params
+                carry no string value under that key for the `Mcp-Name` header.
         """
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         method: str = data["method"]
         opts: CallOptions = {}
         self._stamp(data, opts)
+        # Presence-keyed so the stamp's NAME_BEARING_METHODS rows win by ordering;
+        # a missing/non-string name fails loud rather than omitting a MUST header.
+        headers = opts.setdefault("headers", {})
+        if (key := type(request).name_param) is not None and MCP_NAME_HEADER not in headers:
+            params_data: dict[str, Any] = data.get("params") or {}
+            name = params_data.get(key)
+            if not isinstance(name, str):
+                raise ValueError(f"{method} requires params[{key!r}] for Mcp-Name")
+            headers[MCP_NAME_HEADER] = encode_header_value(name)
         timeout = (
             request_read_timeout_seconds
             if request_read_timeout_seconds is not None
@@ -766,7 +777,7 @@ async def call_tool(
         )
 
         if isinstance(result, types.CallToolResult) and not result.is_error:
-            await self._validate_tool_result(name, result)
+            await self.validate_tool_result(name, result)
 
         if isinstance(result, types.InputRequiredResult) and not allow_input_required:
             raise _input_required_unexpected("call_tool")
@@ -779,8 +790,17 @@ def _resolve_param_headers(self, name: str, arguments: Mapping[str, Any]) -> dic
             return {}
         return mcp_param_headers(header_map, arguments)
 
-    async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
-        """Validate the structured content of a tool result against its output schema."""
+    async def validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
+        """Revalidate a `CallToolResult` against the tool's declared output schema.
+
+        Fetches the tool listing first when `name` has no cached schema. Tools
+        without an output schema (or not listed by the server) pass without
+        validation.
+
+        Raises:
+            RuntimeError: The result's structured content is missing or does
+                not conform to the tool's output schema.
+        """
         if name not in self._tool_output_schemas:
             # refresh output schema cache
             await self.list_tools()
@@ -970,7 +990,7 @@ async def _on_request(
             ctx = ClientRequestContext(
                 session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None
             )
-            response = await self._dispatch_input_request(ctx, request)
+            response = await self.dispatch_input_request(ctx, request)
         client_response = ClientResponse.validate_python(response)
         if isinstance(client_response, types.ErrorData):
             raise MCPError.from_error_data(client_response)
@@ -982,16 +1002,19 @@ async def _on_request(
             raise MCPError(code=INTERNAL_ERROR, message="Client callback returned an invalid result") from None
         return dumped
 
-    async def _dispatch_input_request(
-        self, ctx: ClientRequestContext, req: types.InputRequest
+    async def dispatch_input_request(
+        self, ctx: ClientRequestContext, request: types.InputRequest
     ) -> types.InputResponse | types.ErrorData:
-        """Route a server-initiated input request to the matching constructor callback.
+        """Route an input request through the client's callback table.
 
         Shared by the legacy server→client RPC path (`_on_request`) and the
         2026-07-28 multi-round-trip driver, which dispatches the embedded
         `InputRequiredResult.input_requests` through the same callbacks.
+
+        Returns the callback's `InputResponse`, or `ErrorData` when the
+        callback declines — the refusal path; callers must handle that arm.
         """
-        match req:
+        match request:
             case types.CreateMessageRequest(params=p):
                 return await self._sampling_callback(ctx, p)
             case types.ElicitRequest(params=p):