OpenTelemetry _meta propagation

Author: aabmass

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
     "pyjwt[crypto]>=2.10.1",
     "typing-extensions>=4.13.0",
     "typing-inspection>=0.4.1",
+    "opentelemetry-api>=1.23.0",
 ]
 
 [project.optional-dependencies]
@@ -71,6 +72,7 @@ dev = [
     "coverage[toml]>=7.10.7,<=7.13",
     "pillow>=12.0",
     "strict-no-cover",
+    "opentelemetry-sdk>=1.23.0",
 ]
 docs = [
     "mkdocs>=1.6.1",

src/mcp/shared/session.py

@@ -8,6 +8,7 @@
 
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from opentelemetry.propagate import inject
 from pydantic import BaseModel, TypeAdapter
 from typing_extensions import Self
 
@@ -251,6 +252,9 @@ async def send_request(
         response_stream, response_stream_reader = anyio.create_memory_object_stream[JSONRPCResponse | JSONRPCError](1)
         self._response_streams[request_id] = response_stream
 
+        # Propagate opentelemetry trace context
+        self._inject_otel_context(request)
+
         # Set up progress token if progress callback is provided
         request_data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         if progress_callback is not None:
@@ -295,6 +299,10 @@ async def send_notification(
         related_request_id: RequestId | None = None,
     ) -> None:
         """Emits a notification, which is a one-way message that does not expect a response."""
+
+        # Propagate opentelemetry trace context
+        self._inject_otel_context(notification)
+
         # Some transport implementations may need to set the related_request_id
         # to attribute to the notifications to the request that triggered them.
         jsonrpc_notification = JSONRPCNotification(
@@ -307,6 +315,28 @@ async def send_notification(
         )
         await self._write_stream.send(session_message)
 
+    def _inject_otel_context(self, request: SendRequestT | SendNotificationT) -> None:
+        """Propagate OpenTelemetry context in `_meta`.
+
+        See
+        - SEP414 https://github.com/modelcontextprotocol/modelcontextprotocol/pull/414
+        - OpenTelemetry semantic conventions
+          https://github.com/open-telemetry/semantic-conventions/blob/v1.39.0/docs/gen-ai/mcp.md
+        """
+
+        if request.params is None:
+            return
+
+        carrier: RequestParamsMeta = {}
+        inject(carrier)
+        if not carrier:
+            return
+
+        if request.params.meta is None:
+            request.params.meta = {}
+
+        request.params.meta.update(carrier)
+
     async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:
         if isinstance(response, ErrorData):
             jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=response)