Address review feedback on type annotations and _meta injection

Author: Kludex

src/mcp/server/lowlevel/server.py

@@ -42,7 +42,7 @@ async def main():
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 from importlib.metadata import version as importlib_version
-from typing import Any, Generic
+from typing import Any, Generic, cast
 
 import anyio
 from starlette.applications import Starlette
@@ -451,8 +451,8 @@ async def _handle_request(
         span_name = f"MCP handle {req.method} {target}" if target else f"MCP handle {req.method}"
 
         # Extract W3C trace context from _meta (SEP-414).
-        meta = getattr(req.params, "meta", None) if req.params else None
-        parent_context = extract_trace_context(meta) if isinstance(meta, dict) else None
+        meta = cast(dict[str, Any] | None, getattr(req.params, "meta", None)) if req.params else None
+        parent_context = extract_trace_context(meta) if meta is not None else None
 
         with otel_span(
             span_name,

src/mcp/shared/_otel.py

@@ -27,11 +27,11 @@ def otel_span(
         yield span
 
 
-def inject_trace_context(meta: Any) -> None:
+def inject_trace_context(meta: dict[str, Any]) -> None:
     """Inject W3C trace context (traceparent/tracestate) into a ``_meta`` dict."""
     inject(meta)
 
 
-def extract_trace_context(meta: Any) -> Context:
+def extract_trace_context(meta: dict[str, Any]) -> Context:
     """Extract W3C trace context from a ``_meta`` dict."""
     return extract(meta)

src/mcp/shared/session.py

@@ -278,11 +278,8 @@ async def send_request(
                 attributes={"mcp.method.name": request.method, "jsonrpc.request.id": request_id},
             ):
                 # Inject W3C trace context into _meta (SEP-414).
-                if "params" not in request_data:
-                    request_data["params"] = {}
-                if "_meta" not in request_data["params"]:
-                    request_data["params"]["_meta"] = {}
-                inject_trace_context(request_data["params"]["_meta"])
+                meta: dict[str, Any] = request_data.setdefault("params", {}).setdefault("_meta", {})
+                inject_trace_context(meta)
 
                 jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data)
                 await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata))