feat: JSONRPCDispatcher.send_raw_request emits CLIENT span and injects _meta

Mirrors BaseSession.send_request: outbound requests are wrapped in an otel
CLIENT span and W3C trace context is injected into params._meta (SEP-414).
A side effect is that _meta is always present on the wire (empty under a
no-op tracer), which the interaction suite's sampling/elicitation snapshots
pin.

The contract-test echo recorder now strips _meta so JSON-RPC and direct
dispatch parametrizations record identically.

TODO(maxisbey) marker added: this belongs in an outbound middleware once
that seam exists; the dispatcher should not own otel.

Author: maxisbey

src/mcp/shared/jsonrpc_dispatcher.py

@@ -29,8 +29,10 @@
 import anyio
 import anyio.abc
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from opentelemetry.trace import SpanKind
 from pydantic import ValidationError
 
+from mcp.shared._otel import inject_trace_context, otel_span
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.dispatcher import CallOptions, Dispatcher, OnNotify, OnRequest, ProgressFnT
 from mcp.shared.exceptions import MCPError, NoBackChannelError
@@ -255,17 +257,17 @@ async def send_raw_request(
             raise RuntimeError("JSONRPCDispatcher.send_raw_request called before run() / after close")
         opts = opts or {}
         request_id = self._allocate_id()
-        out_params = dict(params) if params is not None else None
+        out_params = dict(params) if params is not None else {}
+        out_meta = dict(out_params.get("_meta") or {})
         on_progress = opts.get("on_progress")
         if on_progress is not None:
             # The caller wants progress updates. The spec mechanism is: include
             # `_meta.progressToken` on the request; the peer echoes that token on
             # any `notifications/progress` it sends. We use the request id as the
             # token so the receive loop can find this `_Pending.on_progress` by
             # `_pending[token]` without a second lookup table.
-            meta = dict((out_params or {}).get("_meta") or {})
-            meta["progressToken"] = request_id
-            out_params = {**(out_params or {}), "_meta": meta}
+            out_meta["progressToken"] = request_id
+        out_params["_meta"] = out_meta
 
         # buffer=1: at most one outcome is ever delivered. A `WouldBlock` from
         # `_resolve_pending`/`_fan_out_closed` means the waiter already has an
@@ -277,11 +279,26 @@ async def send_raw_request(
         self._pending[request_id] = pending
 
         metadata = _outbound_metadata(_related_request_id, opts)
-        msg = JSONRPCRequest(jsonrpc="2.0", id=request_id, method=method, params=out_params)
+        target = out_params.get("name")
+        span_name = f"MCP send {method}{f' {target}' if isinstance(target, str) else ''}"
+        # TODO(maxisbey): the otel span + inject below mirror
+        # BaseSession.send_request for parity. They belong in an outbound
+        # middleware (symmetric with otel_middleware on the inbound side) once
+        # that seam exists; the dispatcher should not own otel.
         try:
-            await self._write(msg, metadata)
-            with anyio.fail_after(opts.get("timeout")):
-                outcome = await receive.receive()
+            with otel_span(
+                span_name,
+                kind=SpanKind.CLIENT,
+                attributes={"mcp.method.name": method, "jsonrpc.request.id": request_id},
+            ):
+                # Inject W3C trace context into _meta (SEP-414). With a no-op
+                # tracer this writes nothing, but `_meta` itself is still
+                # present on the wire (and the interaction suite pins that).
+                inject_trace_context(out_meta)
+                msg = JSONRPCRequest(jsonrpc="2.0", id=request_id, method=method, params=out_params)
+                await self._write(msg, metadata)
+                with anyio.fail_after(opts.get("timeout")):
+                    outcome = await receive.receive()
         except TimeoutError:
             # Spec-recommended courtesy: tell the peer we've given up so it can
             # stop work and free resources. v1's BaseSession.send_request does

tests/shared/test_dispatcher.py

@@ -34,9 +34,12 @@ def echo_handlers(recorder: Recorder) -> tuple[OnRequest, OnNotify]:
     async def on_request(
         ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
     ) -> dict[str, Any]:
-        recorder.requests.append((method, params))
+        # Strip `_meta` so JSON-RPC and direct dispatch record identically:
+        # the JSON-RPC outbound path always attaches `_meta` (otel injection).
+        recorded = {k: v for k, v in (params or {}).items() if k != "_meta"} if params is not None else None
+        recorder.requests.append((method, recorded))
         recorder.contexts.append(ctx)
-        return {"echoed": method, "params": dict(params or {})}
+        return {"echoed": method, "params": recorded or {}}
 
     async def on_notify(ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None:
         recorder.notifications.append((method, params))

tests/shared/test_jsonrpc_dispatcher.py

@@ -14,7 +14,7 @@
 import pytest
 
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream
-from mcp.shared.dispatcher import DispatchContext
+from mcp.shared.dispatcher import CallOptions, DispatchContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.jsonrpc_dispatcher import (  # pyright: ignore[reportPrivateUsage]
     JSONRPCDispatcher,
@@ -399,6 +399,33 @@ async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] |
     assert received == [(0.25, None, None)]
 
 
+@pytest.mark.anyio
+async def test_send_raw_request_always_carries_meta_on_the_wire():
+    """Outbound requests always include `params._meta` (otel injection per SEP-414).
+
+    Caller-supplied `_meta` keys are preserved; the progress token is merged in.
+    """
+    seen: list[Mapping[str, Any] | None] = []
+
+    async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        seen.append(params)
+        return {}
+
+    async def noop_progress(progress: float, total: float | None, message: str | None) -> None:
+        raise NotImplementedError
+
+    opts: CallOptions = {"on_progress": noop_progress}
+    async with running_pair(jsonrpc_pair, server_on_request=server_on_request) as (client, *_):
+        with anyio.fail_after(5):
+            await client.send_raw_request("a", None)
+            await client.send_raw_request("b", {"x": 1, "_meta": {"k": "v"}}, opts)
+    assert seen[0] == {"_meta": {}}
+    assert seen[1] is not None
+    assert seen[1]["x"] == 1
+    assert seen[1]["_meta"]["k"] == "v"
+    assert "progressToken" in seen[1]["_meta"]
+
+
 @pytest.mark.anyio
 async def test_handler_raising_validation_error_sends_invalid_params():
     async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: