diff --git a/docs/servers/telemetry.mdx b/docs/servers/telemetry.mdx index ed191e6e8..e020cecf4 100644 --- a/docs/servers/telemetry.mdx +++ b/docs/servers/telemetry.mdx @@ -12,7 +12,7 @@ FastMCP includes native OpenTelemetry instrumentation for observability. Traces FastMCP uses the OpenTelemetry API for instrumentation. This means: -- **Zero configuration required** - Instrumentation is always active +- **Zero configuration required** - Native FastMCP instrumentation is available by default - **No overhead when unused** - Without an SDK, all operations are no-ops - **Bring your own SDK** - You control collection, export, and sampling - **Works with any OTEL backend** - Jaeger, Zipkin, Datadog, New Relic, etc. @@ -54,11 +54,11 @@ This works with any OTLP-compatible backend (Jaeger, Zipkin, Grafana Tempo, Data ## Tracing -FastMCP creates spans for all MCP operations, providing end-to-end visibility into request handling. +FastMCP creates spans for the MCP operations it currently instruments, providing end-to-end visibility into request handling. ### Server Spans -The server creates spans for each operation using [MCP semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp/): +The server creates spans for each supported operation using [MCP semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp/): | Span Name | Description | |-----------|-------------| @@ -129,7 +129,7 @@ If another MCP-aware instrumentation layer should own the MCP spans, switch Fast export FASTMCP_TELEMETRY_MODE=propagation_only ``` -In this mode, FastMCP still injects and extracts trace context through MCP `_meta`, but it stops creating its own `tools/call`, `resources/read`, `prompts/get`, and `delegate` spans. +In this mode, FastMCP still injects and extracts trace context through MCP `_meta`, but it stops creating its own MCP request spans and `delegate` spans. Library authors can suppress only FastMCP's native spans programmatically without disabling unrelated nested instrumentation: diff --git a/src/fastmcp/server/http.py b/src/fastmcp/server/http.py index 46d0c956d..bcbc5a69b 100644 --- a/src/fastmcp/server/http.py +++ b/src/fastmcp/server/http.py @@ -12,6 +12,7 @@ from mcp.server.streamable_http import ( EventStore, ) from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from opentelemetry import trace from starlette.applications import Starlette from starlette.middleware import Middleware from starlette.requests import Request @@ -28,6 +29,8 @@ if TYPE_CHECKING: logger = get_logger(__name__) +AMBIENT_SPAN_CONTEXT_SCOPE_KEY = "fastmcp.ambient_span_context" + class StreamableHTTPASGIApp: """ASGI application wrapper for Streamable HTTP server transport.""" @@ -41,6 +44,11 @@ class StreamableHTTPASGIApp: raise RuntimeError( "Task group is not initialized. Make sure to use run()." ) + ambient_span_context = trace.get_current_span().get_span_context() + if ambient_span_context.is_valid: + scope[AMBIENT_SPAN_CONTEXT_SCOPE_KEY] = ambient_span_context + else: + scope.pop(AMBIENT_SPAN_CONTEXT_SCOPE_KEY, None) await self.session_manager.handle_request(scope, receive, send) except RuntimeError as e: if str(e) == "Task group is not initialized. Make sure to use run().": diff --git a/src/fastmcp/server/telemetry.py b/src/fastmcp/server/telemetry.py index d5caf4461..a8374d8d4 100644 --- a/src/fastmcp/server/telemetry.py +++ b/src/fastmcp/server/telemetry.py @@ -7,9 +7,17 @@ from mcp.server.lowlevel.server import request_ctx from opentelemetry import context as otel_context from opentelemetry import trace from opentelemetry.context import Context -from opentelemetry.trace import Link, Span, SpanKind, Status, StatusCode +from opentelemetry.trace import ( + Link, + Span, + SpanContext, + SpanKind, + Status, + StatusCode, +) from fastmcp.exceptions import ToolError as _ToolError +from fastmcp.server.http import AMBIENT_SPAN_CONTEXT_SCOPE_KEY from fastmcp.telemetry import ( extract_trace_context, get_noop_span, @@ -52,7 +60,7 @@ def get_session_span_attributes() -> dict[str, str]: def _get_parent_trace_context() -> tuple[Context | None, list[Link] | None]: """Resolve MCP server parent context plus any ambient transport links.""" - ambient_span_context = trace.get_current_span().get_span_context() + ambient_span_context = _get_ambient_span_context() try: req_ctx = request_ctx.get() @@ -60,10 +68,12 @@ def _get_parent_trace_context() -> tuple[Context | None, list[Link] | None]: meta = dict(req_ctx.meta) if get_trace_context_carrier(meta): parent_context = extract_trace_context(meta) + parent_span_context = trace.get_current_span( + parent_context + ).get_span_context() if ( ambient_span_context.is_valid - and trace.get_current_span(parent_context).get_span_context() - != ambient_span_context + and parent_span_context != ambient_span_context ): return parent_context, [Link(ambient_span_context)] return parent_context, None @@ -76,6 +86,26 @@ def _get_parent_trace_context() -> tuple[Context | None, list[Link] | None]: return None, None +def _get_ambient_span_context() -> SpanContext: + """Resolve the current ambient transport span, if one is available.""" + try: + req_ctx = request_ctx.get() + except LookupError: + req_ctx = None + + if req_ctx is not None: + request = getattr(req_ctx, "request", None) + if request is not None: + ambient_span_context = request.scope.get(AMBIENT_SPAN_CONTEXT_SCOPE_KEY) + if ( + isinstance(ambient_span_context, SpanContext) + and ambient_span_context.is_valid + ): + return ambient_span_context + + return trace.get_current_span().get_span_context() + + @contextmanager def server_span( name: str, @@ -97,12 +127,18 @@ def server_span( parent_context, links = _get_parent_trace_context() tracer = get_tracer() - with tracer.start_as_current_span( + span = tracer.start_span( name, context=parent_context, kind=SpanKind.SERVER, links=links, - ) as span: + ) + current_context = trace.set_span_in_context( + span, + parent_context if parent_context is not None else otel_context.get_current(), + ) + token = otel_context.attach(current_context) + try: if span.is_recording(): attrs: dict[str, str] = { # MCP semantic conventions @@ -132,6 +168,9 @@ def server_span( span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) raise + finally: + otel_context.detach(token) + span.end() @contextmanager diff --git a/src/fastmcp/telemetry.py b/src/fastmcp/telemetry.py index 386808afe..b8c5c22a3 100644 --- a/src/fastmcp/telemetry.py +++ b/src/fastmcp/telemetry.py @@ -145,12 +145,13 @@ def extract_trace_context(meta: dict[str, Any] | None) -> Context: meta: The meta dict from an MCP request (ctx.request_context.meta) Returns: - An OpenTelemetry Context with the extracted trace context, - or the current context if no trace context was propagated + An OpenTelemetry Context with propagated trace context and baggage + merged onto the current context, or the current context if no + propagation keys were present. """ carrier = get_trace_context_carrier(meta) if carrier: - return propagate.extract(carrier) + return propagate.extract(carrier, context=otel_context.get_current()) return otel_context.get_current() diff --git a/tests/telemetry/test_interop.py b/tests/telemetry/test_interop.py index b547b62d1..aab2d6784 100644 --- a/tests/telemetry/test_interop.py +++ b/tests/telemetry/test_interop.py @@ -2,12 +2,15 @@ from __future__ import annotations +import json from typing import Any, cast -from opentelemetry import context as otel_context -from opentelemetry import trace -from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +import httpx from mcp.server.lowlevel.server import request_ctx +from opentelemetry import baggage, trace +from opentelemetry import context as otel_context +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from starlette.middleware import Middleware from fastmcp import Client, FastMCP from fastmcp.client.telemetry import client_span @@ -21,6 +24,31 @@ class DummyReqCtx: def __init__(self, meta: dict[str, str]): self.meta = meta + self.request = None + + +class AmbientHTTPSpanMiddleware: + """Minimal ASGI middleware that simulates outer HTTP instrumentation.""" + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + tracer = trace.get_tracer("ambient-http") + with tracer.start_as_current_span("ambient-http-request"): + await self.app(scope, receive, send) + + +def parse_sse_response(body: str) -> dict[str, Any]: + """Extract the first SSE data payload from a streamable HTTP response.""" + for line in body.splitlines(): + if line.startswith("data: "): + return json.loads(line[6:]) + raise AssertionError(f"Missing SSE data payload in response: {body!r}") class TestClientInteropMode: @@ -93,6 +121,209 @@ class TestClientInteropMode: class TestServerInteropMode: + async def test_streamable_http_third_party_client_uses_meta_parent_and_baggage( + self, + trace_exporter: InMemorySpanExporter, + ): + child = FastMCP("child-server") + + @child.tool() + def tenant() -> str: + tenant_name = baggage.get_baggage("tenant") + return tenant_name if isinstance(tenant_name, str) else "missing" + + parent = FastMCP("parent-server") + parent.mount(child, namespace="child") + + app = parent.http_app( + transport="http", + path="/mcp", + middleware=[Middleware(AmbientHTTPSpanMiddleware)], + ) + headers = { + "accept": "application/json, text/event-stream", + "content-type": "application/json", + } + + async with app.router.lifespan_context(app): + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url="http://testserver", + ) as client: + init_response = await client.post( + "/mcp", + headers=headers, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "clientInfo": { + "name": "opaque-client", + "version": "0.1.0", + }, + }, + }, + ) + session_id = init_response.headers["mcp-session-id"] + await client.post( + "/mcp", + headers={**headers, "mcp-session-id": session_id}, + json={ + "jsonrpc": "2.0", + "method": "notifications/initialized", + "params": {}, + }, + ) + + trace_exporter.clear() + tracer = trace.get_tracer("opaque-client") + baggage_token = otel_context.attach( + baggage.set_baggage("tenant", "acme") + ) + try: + with tracer.start_as_current_span( + "opaque-client-root" + ) as client_span_export: + meta = inject_trace_context() + tool_response = await client.post( + "/mcp", + headers={**headers, "mcp-session-id": session_id}, + json={ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "child_tenant", + "arguments": {}, + "_meta": meta, + }, + }, + ) + finally: + otel_context.detach(baggage_token) + + payload = parse_sse_response(tool_response.text) + assert payload["result"]["content"][0]["text"] == "acme" + + spans = { + span.name: span + for span in trace_exporter.get_finished_spans() + if span.name + in { + "ambient-http-request", + "opaque-client-root", + "tools/call child_tenant", + "delegate tenant", + "tools/call tenant", + } + } + parent_span_export = spans["tools/call child_tenant"] + delegate_span_export = spans["delegate tenant"] + child_span_export = spans["tools/call tenant"] + ambient_span_export = spans["ambient-http-request"] + + assert parent_span_export.parent is not None + assert ( + parent_span_export.parent.span_id + == client_span_export.get_span_context().span_id + ) + assert any( + link.context.span_id == ambient_span_export.get_span_context().span_id + for link in parent_span_export.links + ) + assert child_span_export.parent is not None + assert ( + child_span_export.parent.span_id + == client_span_export.get_span_context().span_id + ) + assert delegate_span_export.parent is not None + assert ( + delegate_span_export.parent.span_id + == parent_span_export.get_span_context().span_id + ) + assert any( + link.context.span_id == ambient_span_export.get_span_context().span_id + for link in child_span_export.links + ) + + async def test_server_span_makes_propagated_baggage_current( + self, + monkeypatch, + trace_exporter: InMemorySpanExporter, + ): + import fastmcp.server.telemetry as server_telemetry + + monkeypatch.setattr(server_telemetry, "get_auth_span_attributes", lambda: {}) + monkeypatch.setattr(server_telemetry, "get_session_span_attributes", lambda: {}) + + tracer = trace.get_tracer("external") + baggage_token = otel_context.attach(baggage.set_baggage("tenant", "acme")) + try: + with tracer.start_as_current_span("external-client-parent"): + meta = inject_trace_context() + finally: + otel_context.detach(baggage_token) + + req_token = request_ctx.set(cast(Any, DummyReqCtx(meta or {}))) + try: + with server_span( + "tools/call weather", + "tools/call", + "test-server", + "tool", + "weather", + tool_name="weather", + ): + assert baggage.get_baggage("tenant") == "acme" + finally: + request_ctx.reset(req_token) + + async def test_server_span_with_baggage_only_meta_keeps_ambient_parent( + self, + monkeypatch, + trace_exporter: InMemorySpanExporter, + ): + import fastmcp.server.telemetry as server_telemetry + + monkeypatch.setattr(server_telemetry, "get_auth_span_attributes", lambda: {}) + monkeypatch.setattr(server_telemetry, "get_session_span_attributes", lambda: {}) + + with trace.get_tracer("external").start_as_current_span( + "ambient-http-request" + ) as ambient_span: + req_token = request_ctx.set( + cast(Any, DummyReqCtx({"baggage": "userId=alice"})) + ) + try: + with server_span( + "tools/call weather", + "tools/call", + "test-server", + "tool", + "weather", + tool_name="weather", + ): + pass + finally: + request_ctx.reset(req_token) + + spans = { + span.name: span + for span in trace_exporter.get_finished_spans() + if span.name in {"ambient-http-request", "tools/call weather"} + } + server_span_export = spans["tools/call weather"] + + assert server_span_export.parent is not None + assert ( + server_span_export.parent.span_id == ambient_span.get_span_context().span_id + ) + assert server_span_export.links == () + async def test_server_span_uses_meta_parent_and_links_ambient_context( self, monkeypatch, @@ -138,7 +369,10 @@ class TestServerInteropMode: server_span_export = spans["tools/call weather"] assert server_span_export.parent is not None - assert server_span_export.parent.span_id == remote_parent.get_span_context().span_id + assert ( + server_span_export.parent.span_id + == remote_parent.get_span_context().span_id + ) assert any( link.context.span_id == ambient_span.get_span_context().span_id for link in server_span_export.links diff --git a/tests/telemetry/test_module.py b/tests/telemetry/test_module.py index d434ed51c..e66688469 100644 --- a/tests/telemetry/test_module.py +++ b/tests/telemetry/test_module.py @@ -104,6 +104,18 @@ class TestExtractTraceContext: assert span_ctx.is_valid assert format(span_ctx.trace_id, "032x") == "0af7651916cd43dd8448eb211c80319c" + def test_baggage_only_meta_preserves_current_span( + self, trace_exporter: InMemorySpanExporter + ): + tracer = get_tracer() + with tracer.start_as_current_span("current") as current_span: + ctx = extract_trace_context({"baggage": "userId=alice"}) + + span_ctx = trace.get_current_span(ctx).get_span_context() + assert span_ctx.is_valid + assert span_ctx.span_id == current_span.get_span_context().span_id + assert baggage.get_baggage("userId", context=ctx) == "alice" + def test_none_meta_returns_current_context( self, trace_exporter: InMemorySpanExporter ):