From 2e6e8256d9b0865bf8ce93797be6480be0651685 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 4 Oct 2025 02:23:43 +0000 Subject: [PATCH 1/3] Document OpenTelemetry integration Add comprehensive documentation for integrating OpenTelemetry with FastMCP: - New integration guide at docs/integrations/opentelemetry.mdx - Covers logging integration with LoggingHandler - Demonstrates span creation via custom middleware - Includes production OTLP export configuration - Provides complete working example Also update related docs: - Add OpenTelemetry reference in middleware.mdx - Add tip about OpenTelemetry in logging.mdx - Register doc in docs.json under new Observability section Example code: - examples/opentelemetry_example.py with working weather server Closes #1998 Co-authored-by: William Easton --- docs/docs.json | 7 + docs/integrations/opentelemetry.mdx | 517 ++++++++++++++++++++++++++++ docs/servers/logging.mdx | 2 + docs/servers/middleware.mdx | 6 +- examples/opentelemetry_example.py | 234 +++++++++++++ 5 files changed, 765 insertions(+), 1 deletion(-) create mode 100644 docs/integrations/opentelemetry.mdx create mode 100644 examples/opentelemetry_example.py diff --git a/docs/docs.json b/docs/docs.json index 758eca70f..3568b8745 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -194,6 +194,13 @@ "integrations/permit" ] }, + { + "group": "Observability", + "icon": "chart-line", + "pages": [ + "integrations/opentelemetry" + ] + }, { "group": "AI Assistants", "icon": "robot", diff --git a/docs/integrations/opentelemetry.mdx b/docs/integrations/opentelemetry.mdx new file mode 100644 index 000000000..431fd412d --- /dev/null +++ b/docs/integrations/opentelemetry.mdx @@ -0,0 +1,517 @@ +--- +title: OpenTelemetry Integration +description: Instrument your FastMCP server with OpenTelemetry for distributed tracing and observability +icon: chart-line +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + +OpenTelemetry provides comprehensive observability for your FastMCP servers through distributed tracing, logging, and metrics. FastMCP's existing logging infrastructure and middleware system integrate seamlessly with OpenTelemetry without requiring any changes to FastMCP itself. + +## Why OpenTelemetry? + +OpenTelemetry is the industry-standard observability framework that provides: + +- **Distributed Tracing**: Track MCP operations across your system with spans +- **Structured Logging**: Export FastMCP logs to observability backends +- **Metrics Collection**: Monitor performance and usage patterns +- **Vendor Agnostic**: Works with Jaeger, Zipkin, Grafana, Datadog, and more +- **Production Ready**: Battle-tested with stable APIs for tracing and metrics + +## Prerequisites + +Install OpenTelemetry packages for Python: + +```bash +pip install opentelemetry-api opentelemetry-sdk +``` + +For production deployments with OTLP export: + +```bash +pip install opentelemetry-exporter-otlp-proto-grpc +``` + + +OpenTelemetry supports Python 3.9 and higher. Tracing and metrics are stable, while logging is in active development. + + +## Logging Integration + +FastMCP uses Python's standard `logging` module, which OpenTelemetry can instrument directly using `LoggingHandler`. This sends your FastMCP logs to any OpenTelemetry-compatible backend. + +### Basic Setup + +```python +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter +from opentelemetry._logs import set_logger_provider +from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogExporter + +from fastmcp import FastMCP +from fastmcp.utilities.logging import get_logger + +# Configure OpenTelemetry +resource = Resource(attributes={ + "service.name": "my-fastmcp-server", + "service.version": "1.0.0", +}) + +# Set up tracing +trace_provider = TracerProvider(resource=resource) +trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) +trace.set_tracer_provider(trace_provider) + +# Set up logging +logger_provider = LoggerProvider(resource=resource) +logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) +set_logger_provider(logger_provider) + +# Attach OpenTelemetry to FastMCP's logger +fastmcp_logger = get_logger("my_server") +fastmcp_logger.addHandler(LoggingHandler(logger_provider=logger_provider)) + +# Create your FastMCP server +mcp = FastMCP("My Server") + +@mcp.tool() +def greet(name: str) -> str: + """Greet someone by name.""" + fastmcp_logger.info(f"Greeting {name}") + return f"Hello, {name}!" +``` + +### Production OTLP Export + +For production environments, replace console exporters with OTLP exporters: + +```python +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter + +# Configure OTLP endpoint (e.g., Grafana, Jaeger, or any OTLP collector) +otlp_endpoint = "http://localhost:4317" + +# Tracing +trace_provider = TracerProvider(resource=resource) +trace_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint)) +) +trace.set_tracer_provider(trace_provider) + +# Logging +logger_provider = LoggerProvider(resource=resource) +logger_provider.add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter(endpoint=otlp_endpoint)) +) +set_logger_provider(logger_provider) +``` + +### Structured Logging with OpenTelemetry + +FastMCP's `StructuredLoggingMiddleware` outputs JSON logs that OpenTelemetry collectors can parse and enrich: + +```python +from fastmcp import FastMCP +from fastmcp.server.middleware.logging import StructuredLoggingMiddleware + +mcp = FastMCP("Structured Server") + +# Add structured logging middleware +mcp.add_middleware(StructuredLoggingMiddleware( + include_payloads=True, + max_payload_length=1000 +)) + +# OpenTelemetry will capture these structured logs +``` + +The structured logs include metadata like request timestamps, method names, token estimates, and payload sizes - perfect for observability platforms. + +## Spans via Middleware + +FastMCP's middleware system provides the perfect foundation for creating OpenTelemetry spans that track MCP operations. + +### Basic Tracing Middleware + +Create a middleware that emits spans for all MCP requests: + +```python +from opentelemetry import trace +from opentelemetry.trace import Status, StatusCode +from fastmcp.server.middleware import Middleware, MiddlewareContext + +class OpenTelemetryMiddleware(Middleware): + """Middleware that creates OpenTelemetry spans for MCP operations.""" + + def __init__(self, tracer_name: str = "fastmcp"): + self.tracer = trace.get_tracer(tracer_name) + + async def on_request(self, context: MiddlewareContext, call_next): + """Create a span for each MCP request.""" + with self.tracer.start_as_current_span( + f"mcp.{context.method}", + attributes={ + "mcp.method": context.method, + "mcp.source": context.source, + "mcp.type": context.type, + } + ) as span: + try: + result = await call_next(context) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + +# Add to your server +mcp.add_middleware(OpenTelemetryMiddleware()) +``` + +### Tool-Specific Spans + +For more granular tracing, create spans specifically for tool executions: + +```python +from opentelemetry import trace +from opentelemetry.trace import Status, StatusCode +from fastmcp.server.middleware import Middleware, MiddlewareContext + +class ToolTracingMiddleware(Middleware): + """Create detailed spans for tool executions.""" + + def __init__(self, tracer_name: str = "fastmcp.tools"): + self.tracer = trace.get_tracer(tracer_name) + + async def on_call_tool(self, context: MiddlewareContext, call_next): + """Create a span for each tool call with detailed attributes.""" + tool_name = context.message.name + + with self.tracer.start_as_current_span( + f"tool.{tool_name}", + attributes={ + "mcp.tool.name": tool_name, + "mcp.tool.arguments": str(context.message.arguments), + } + ) as span: + try: + result = await call_next(context) + + # Add result metadata to span + span.set_attribute("mcp.tool.success", True) + if hasattr(result, 'content'): + span.set_attribute("mcp.tool.content_length", len(str(result.content))) + + span.set_status(Status(StatusCode.OK)) + return result + + except Exception as e: + span.set_attribute("mcp.tool.success", False) + span.set_attribute("mcp.tool.error", str(e)) + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + +mcp.add_middleware(ToolTracingMiddleware()) +``` + +### Comprehensive Observability Middleware + +For production systems, create a middleware that handles all MCP operation types: + +```python +from opentelemetry import trace +from opentelemetry.trace import Status, StatusCode +from fastmcp.server.middleware import Middleware, MiddlewareContext + +class ComprehensiveTracingMiddleware(Middleware): + """Complete tracing for tools, resources, and prompts.""" + + def __init__(self, tracer_name: str = "fastmcp"): + self.tracer = trace.get_tracer(tracer_name) + + async def on_call_tool(self, context: MiddlewareContext, call_next): + """Trace tool executions.""" + return await self._trace_operation( + "tool.call", + {"tool.name": context.message.name}, + context, + call_next + ) + + async def on_read_resource(self, context: MiddlewareContext, call_next): + """Trace resource reads.""" + return await self._trace_operation( + "resource.read", + {"resource.uri": context.message.uri}, + context, + call_next + ) + + async def on_get_prompt(self, context: MiddlewareContext, call_next): + """Trace prompt retrievals.""" + return await self._trace_operation( + "prompt.get", + {"prompt.name": context.message.name}, + context, + call_next + ) + + async def _trace_operation( + self, + operation_name: str, + attributes: dict, + context: MiddlewareContext, + call_next + ): + """Helper to create spans with consistent attributes.""" + with self.tracer.start_as_current_span( + operation_name, + attributes={ + "mcp.method": context.method, + "mcp.source": context.source, + **attributes, + } + ) as span: + try: + result = await call_next(context) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + +mcp.add_middleware(ComprehensiveTracingMiddleware()) +``` + +## Complete Example + +Here's a production-ready example combining logging and tracing: + +```python +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter +from opentelemetry._logs import set_logger_provider +from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogExporter + +from fastmcp import FastMCP +from fastmcp.utilities.logging import get_logger +from fastmcp.server.middleware import Middleware, MiddlewareContext +from opentelemetry.trace import Status, StatusCode + +# Configure OpenTelemetry +resource = Resource(attributes={ + "service.name": "weather-mcp-server", + "service.version": "1.0.0", + "deployment.environment": "production", +}) + +# Tracing setup +trace_provider = TracerProvider(resource=resource) +trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) +trace.set_tracer_provider(trace_provider) + +# Logging setup +logger_provider = LoggerProvider(resource=resource) +logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) +set_logger_provider(logger_provider) + +# Middleware for tracing +class TracingMiddleware(Middleware): + def __init__(self): + self.tracer = trace.get_tracer("weather-server") + + async def on_call_tool(self, context: MiddlewareContext, call_next): + with self.tracer.start_as_current_span( + f"tool.{context.message.name}", + attributes={"tool.name": context.message.name} + ) as span: + try: + result = await call_next(context) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + +# Create FastMCP server +mcp = FastMCP("Weather Server") + +# Attach OpenTelemetry to FastMCP logger +logger = get_logger("weather") +logger.addHandler(LoggingHandler(logger_provider=logger_provider)) + +# Add tracing middleware +mcp.add_middleware(TracingMiddleware()) + +@mcp.tool() +def get_weather(city: str) -> dict: + """Get weather for a city.""" + logger.info(f"Fetching weather for {city}") + return {"city": city, "temp": 72, "condition": "sunny"} + +if __name__ == "__main__": + mcp.run() +``` + +## Exporting to Observability Backends + +### Console Exporter (Development) + +The console exporter is perfect for local development and testing: + +```python +from opentelemetry.sdk.trace.export import ConsoleSpanExporter +from opentelemetry.sdk._logs.export import ConsoleLogExporter + +# Already shown in examples above +trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) +logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) +``` + +### OTLP Exporter (Production) + +OTLP (OpenTelemetry Protocol) works with most modern observability platforms: + +```python +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter + +# Configure for your backend +otlp_endpoint = "http://your-collector:4317" + +trace_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint)) +) +logger_provider.add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter(endpoint=otlp_endpoint)) +) +``` + +Supported backends include: +- **Grafana** with Tempo and Loki +- **Jaeger** for distributed tracing +- **Zipkin** for trace visualization +- **Datadog**, **New Relic**, **Honeycomb** (commercial platforms) +- **Self-hosted** OpenTelemetry Collector + +### Environment Variables + +OpenTelemetry exporters can be configured via environment variables: + +```bash +export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317" +export OTEL_SERVICE_NAME="my-fastmcp-server" +export OTEL_RESOURCE_ATTRIBUTES="deployment.environment=production" +``` + +Then in your code: + +```python +# OpenTelemetry will automatically use environment variables +trace_provider = TracerProvider() +trace_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter()) # Uses OTEL_EXPORTER_OTLP_ENDPOINT +) +``` + +## Best Practices + +### When to Use Logging vs Spans + +- **Logging**: Discrete events, errors, diagnostic messages +- **Spans**: Operations with duration, distributed tracing across services + +For FastMCP servers: +- Use **spans** for tool calls, resource reads, prompt executions +- Use **logging** for validation errors, configuration issues, business logic events + +### Performance Considerations + +OpenTelemetry is designed for production, but follow these guidelines: + +1. **Use BatchProcessors**: Always use `BatchSpanProcessor` and `BatchLogRecordProcessor` rather than synchronous exporters +2. **Sampling**: For high-volume servers, configure sampling to reduce overhead: + +```python +from opentelemetry.sdk.trace.sampling import TraceIdRatioBased + +# Sample 10% of traces +trace_provider = TracerProvider( + resource=resource, + sampler=TraceIdRatioBased(0.1) +) +``` + +3. **Attribute Limits**: Avoid adding large payloads as span attributes. Use `max_payload_length` in middleware: + +```python +# Good - limit attribute size +span.set_attribute("tool.arguments", str(args)[:500]) + +# Bad - unbounded attribute size +span.set_attribute("tool.arguments", str(args)) # Could be huge! +``` + +### Security: Avoiding Sensitive Data + +Never log sensitive information in traces or logs: + +```python +async def on_call_tool(self, context: MiddlewareContext, call_next): + tool_name = context.message.name + + # Redact sensitive arguments + safe_args = { + k: v if k not in ["password", "api_key", "token"] else "***REDACTED***" + for k, v in context.message.arguments.items() + } + + with self.tracer.start_as_current_span( + f"tool.{tool_name}", + attributes={"tool.arguments": str(safe_args)} + ) as span: + return await call_next(context) +``` + +### Integration with FastMCP Middleware + +OpenTelemetry middleware works seamlessly with FastMCP's built-in middleware: + +```python +from fastmcp.server.middleware.timing import TimingMiddleware +from fastmcp.server.middleware.logging import LoggingMiddleware + +# Order matters: error handling first, then tracing, then logging +mcp.add_middleware(ErrorHandlingMiddleware()) +mcp.add_middleware(OpenTelemetryMiddleware()) # Your custom middleware +mcp.add_middleware(TimingMiddleware()) # Built-in timing +mcp.add_middleware(LoggingMiddleware()) # Built-in logging +``` + +The execution order ensures: +1. Errors are handled consistently +2. OpenTelemetry captures complete request lifecycle +3. Timing data is included in spans +4. Everything is logged with proper context + +## Additional Resources + +- [OpenTelemetry Python Documentation](https://opentelemetry.io/docs/languages/python/) +- [FastMCP Middleware Guide](/servers/middleware) +- [FastMCP Logging Guide](/servers/logging) +- [OpenTelemetry Semantic Conventions](https://opentelemetry.io/docs/specs/semconv/) + + +For examples and sample code, see [`examples/opentelemetry_example.py`](https://github.com/jlowin/fastmcp/tree/main/examples/opentelemetry_example.py) in the FastMCP repository. + diff --git a/docs/servers/logging.mdx b/docs/servers/logging.mdx index 6e27aa72d..7f44e35db 100644 --- a/docs/servers/logging.mdx +++ b/docs/servers/logging.mdx @@ -9,6 +9,8 @@ import { VersionBadge } from '/snippets/version-badge.mdx' This documentation covers **MCP client logging** - sending messages from your server to MCP clients. For standard server-side logging (e.g., writing to files, console), use `fastmcp.utilities.logging.get_logger()` or Python's built-in `logging` module. + +For production observability and distributed tracing, see the [OpenTelemetry Integration](/integrations/opentelemetry) guide. Server logging allows MCP tools to send debug, info, warning, and error messages back to the client. This provides visibility into function execution and helps with debugging during development and operation. diff --git a/docs/servers/middleware.mdx b/docs/servers/middleware.mdx index 5ef8d8fe1..72b213788 100644 --- a/docs/servers/middleware.mdx +++ b/docs/servers/middleware.mdx @@ -444,7 +444,11 @@ The built-in versions include custom logger support, proper formatting, and **De ### Logging Middleware -Request and response logging is crucial for debugging, monitoring, and understanding usage patterns in your MCP server. FastMCP provides comprehensive logging middleware at `fastmcp.server.middleware.logging`. +Request and response logging is crucial for debugging, monitoring, and understanding usage patterns in your MCP server. FastMCP provides comprehensive logging middleware at `fastmcp.server.middleware.logging`. + + +For production observability with distributed tracing, see the [OpenTelemetry Integration](/integrations/opentelemetry) guide for instrumenting your server with OpenTelemetry spans and logging. + Here's an example of how it works: diff --git a/examples/opentelemetry_example.py b/examples/opentelemetry_example.py new file mode 100644 index 000000000..9d21c4649 --- /dev/null +++ b/examples/opentelemetry_example.py @@ -0,0 +1,234 @@ +""" +OpenTelemetry Integration Example + +This example demonstrates how to integrate OpenTelemetry with FastMCP for +comprehensive observability. It shows: + +1. Configuring OpenTelemetry tracing and logging +2. Creating custom middleware that emits spans +3. Attaching OpenTelemetry to FastMCP's logger +4. Exporting to console (easily switch to OTLP for production) + +To run this example: + uv run examples/opentelemetry_example.py + +For production, replace ConsoleSpanExporter/ConsoleLogExporter with: + from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter + from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter + +Requirements: + pip install opentelemetry-api opentelemetry-sdk +""" + +from opentelemetry import trace +from opentelemetry._logs import set_logger_provider +from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogExporter +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter +from opentelemetry.trace import Status, StatusCode + +from fastmcp import FastMCP +from fastmcp.server.middleware import Middleware, MiddlewareContext +from fastmcp.utilities.logging import get_logger + +# ============================================================================ +# OpenTelemetry Configuration +# ============================================================================ + +# Define service metadata +resource = Resource( + attributes={ + "service.name": "fastmcp-weather-server", + "service.version": "1.0.0", + "deployment.environment": "development", + } +) + +# Configure tracing +trace_provider = TracerProvider(resource=resource) +trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) +trace.set_tracer_provider(trace_provider) + +# Configure logging +logger_provider = LoggerProvider(resource=resource) +logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) +set_logger_provider(logger_provider) + +# ============================================================================ +# Custom Middleware for OpenTelemetry Spans +# ============================================================================ + + +class OpenTelemetryMiddleware(Middleware): + """Middleware that creates OpenTelemetry spans for MCP operations.""" + + def __init__(self, tracer_name: str = "fastmcp"): + self.tracer = trace.get_tracer(tracer_name) + + async def on_call_tool(self, context: MiddlewareContext, call_next): + """Create a span for each tool call with detailed attributes.""" + tool_name = context.message.name + + # Create a span for this tool call + with self.tracer.start_as_current_span( + f"tool.{tool_name}", + attributes={ + "mcp.method": context.method, + "mcp.source": context.source, + "mcp.tool.name": tool_name, + "mcp.tool.arguments": str(context.message.arguments), + }, + ) as span: + try: + # Execute the tool + result = await call_next(context) + + # Mark span as successful + span.set_attribute("mcp.tool.success", True) + span.set_status(Status(StatusCode.OK)) + + return result + + except Exception as e: + # Record the error in the span + span.set_attribute("mcp.tool.success", False) + span.set_attribute("mcp.tool.error", str(e)) + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + +# ============================================================================ +# FastMCP Server Setup +# ============================================================================ + +# Create FastMCP server +mcp = FastMCP("Weather Server") + +# Attach OpenTelemetry to FastMCP's logger +logger = get_logger("weather") +logger.addHandler(LoggingHandler(logger_provider=logger_provider)) + +# Add OpenTelemetry middleware +mcp.add_middleware(OpenTelemetryMiddleware()) + +# ============================================================================ +# Server Tools +# ============================================================================ + + +@mcp.tool() +def get_weather(city: str) -> dict: + """Get current weather for a city. + + Args: + city: Name of the city + + Returns: + Weather information including temperature and conditions + """ + logger.info(f"Fetching weather for {city}") + + # Simulate weather lookup + weather_data = { + "city": city, + "temperature": 72, + "condition": "sunny", + "humidity": 45, + } + + logger.info( + f"Weather retrieved: {weather_data['condition']}, {weather_data['temperature']}°F" + ) + + return weather_data + + +@mcp.tool() +def get_forecast(city: str, days: int = 3) -> dict: + """Get weather forecast for a city. + + Args: + city: Name of the city + days: Number of days to forecast (1-7) + + Returns: + Forecast data for the specified number of days + """ + logger.info(f"Fetching {days}-day forecast for {city}") + + if days < 1 or days > 7: + logger.warning(f"Invalid days parameter: {days}. Must be 1-7.") + raise ValueError("Days must be between 1 and 7") + + # Simulate forecast data + forecast = { + "city": city, + "days": days, + "forecast": [ + {"day": i + 1, "temp": 70 + i, "condition": "partly cloudy"} + for i in range(days) + ], + } + + logger.info(f"Forecast retrieved for {days} days") + + return forecast + + +@mcp.tool() +def convert_temperature(temp: float, from_unit: str, to_unit: str) -> dict: + """Convert temperature between Fahrenheit and Celsius. + + Args: + temp: Temperature value to convert + from_unit: Source unit ('F' or 'C') + to_unit: Target unit ('F' or 'C') + + Returns: + Converted temperature value + """ + logger.debug(f"Converting {temp}°{from_unit} to °{to_unit}") + + # Validate units + if from_unit not in ["F", "C"] or to_unit not in ["F", "C"]: + logger.error(f"Invalid units: {from_unit} or {to_unit}") + raise ValueError("Units must be 'F' or 'C'") + + # Perform conversion + if from_unit == to_unit: + result = temp + elif from_unit == "F" and to_unit == "C": + result = (temp - 32) * 5 / 9 + else: # from_unit == "C" and to_unit == "F" + result = (temp * 9 / 5) + 32 + + logger.info(f"Converted {temp}°{from_unit} to {result:.1f}°{to_unit}") + + return { + "original": {"value": temp, "unit": from_unit}, + "converted": {"value": round(result, 1), "unit": to_unit}, + } + + +# ============================================================================ +# Main +# ============================================================================ + +if __name__ == "__main__": + print("=" * 70) + print("FastMCP + OpenTelemetry Example") + print("=" * 70) + print("\nThis example demonstrates OpenTelemetry integration with FastMCP.") + print("Watch the console for:") + print(" - Trace spans showing tool execution timing") + print(" - Log entries from FastMCP's logger") + print("\nFor production, replace console exporters with OTLP exporters") + print("to send data to Grafana, Jaeger, or other observability platforms.") + print("=" * 70) + print() + + # Run the server + mcp.run() From 8ecce7a95c89e1c0b916d56320101ba0719c10a0 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Wed, 3 Dec 2025 22:46:14 +0000 Subject: [PATCH 2/3] Add built-in OpenTelemetry instrumentation middleware - Create OpenTelemetryMiddleware with automatic span creation for all MCP operations - Add opentelemetry as optional dependency (pip install fastmcp[opentelemetry]) - Gracefully degrades to no-op when OpenTelemetry not installed - Configuration options for privacy (include_arguments) and performance (max_argument_length) - Update documentation to reflect built-in support - Update example to use built-in middleware - Add comprehensive tests Co-authored-by: William Easton --- docs/integrations/opentelemetry.mdx | 259 +++++---------- examples/opentelemetry_example.py | 47 +-- pyproject.toml | 6 +- .../server/middleware/opentelemetry.py | 313 ++++++++++++++++++ .../test_opentelemetry_middleware.py | 180 ++++++++++ uv.lock | 73 +++- 6 files changed, 659 insertions(+), 219 deletions(-) create mode 100644 src/fastmcp/server/middleware/opentelemetry.py create mode 100644 tests/server/middleware/test_opentelemetry_middleware.py diff --git a/docs/integrations/opentelemetry.mdx b/docs/integrations/opentelemetry.mdx index 431fd412d..6a27e1dea 100644 --- a/docs/integrations/opentelemetry.mdx +++ b/docs/integrations/opentelemetry.mdx @@ -6,7 +6,7 @@ icon: chart-line import { VersionBadge } from "/snippets/version-badge.mdx" -OpenTelemetry provides comprehensive observability for your FastMCP servers through distributed tracing, logging, and metrics. FastMCP's existing logging infrastructure and middleware system integrate seamlessly with OpenTelemetry without requiring any changes to FastMCP itself. +FastMCP includes built-in OpenTelemetry instrumentation that automatically creates spans for all MCP operations. The integration provides comprehensive observability through distributed tracing, logging, and metrics with zero configuration required. ## Why OpenTelemetry? @@ -18,9 +18,29 @@ OpenTelemetry is the industry-standard observability framework that provides: - **Vendor Agnostic**: Works with Jaeger, Zipkin, Grafana, Datadog, and more - **Production Ready**: Battle-tested with stable APIs for tracing and metrics -## Prerequisites +## Quick Start -Install OpenTelemetry packages for Python: +FastMCP includes OpenTelemetry middleware out of the box. Simply add the middleware to your server: + +```python +from fastmcp import FastMCP +from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + +mcp = FastMCP("My Server") +mcp.add_middleware(OpenTelemetryMiddleware()) + +@mcp.tool() +def greet(name: str) -> str: + return f"Hello, {name}!" +``` + +If you don't have OpenTelemetry installed, the middleware gracefully becomes a no-op. To enable full instrumentation: + +```bash +pip install fastmcp[opentelemetry] +``` + +Or install the packages directly: ```bash pip install opentelemetry-api opentelemetry-sdk @@ -131,163 +151,84 @@ mcp.add_middleware(StructuredLoggingMiddleware( The structured logs include metadata like request timestamps, method names, token estimates, and payload sizes - perfect for observability platforms. -## Spans via Middleware +## Built-in Tracing Middleware -FastMCP's middleware system provides the perfect foundation for creating OpenTelemetry spans that track MCP operations. +FastMCP includes `OpenTelemetryMiddleware` that automatically creates spans for all MCP operations including tools, resources, prompts, and list operations. -### Basic Tracing Middleware +### Configuration Options -Create a middleware that emits spans for all MCP requests: +The middleware supports several configuration options: ```python -from opentelemetry import trace -from opentelemetry.trace import Status, StatusCode -from fastmcp.server.middleware import Middleware, MiddlewareContext +from fastmcp import FastMCP +from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware -class OpenTelemetryMiddleware(Middleware): - """Middleware that creates OpenTelemetry spans for MCP operations.""" +mcp = FastMCP("My Server") - def __init__(self, tracer_name: str = "fastmcp"): - self.tracer = trace.get_tracer(tracer_name) - - async def on_request(self, context: MiddlewareContext, call_next): - """Create a span for each MCP request.""" - with self.tracer.start_as_current_span( - f"mcp.{context.method}", - attributes={ - "mcp.method": context.method, - "mcp.source": context.source, - "mcp.type": context.type, - } - ) as span: - try: - result = await call_next(context) - span.set_status(Status(StatusCode.OK)) - return result - except Exception as e: - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - raise - -# Add to your server +# Default configuration (recommended) mcp.add_middleware(OpenTelemetryMiddleware()) + +# Custom configuration +mcp.add_middleware(OpenTelemetryMiddleware( + tracer_name="my-custom-tracer", # Custom tracer name + enabled=True, # Explicitly enable/disable + include_arguments=False, # Don't include arguments for privacy + max_argument_length=1000 # Limit argument string length in spans +)) ``` -### Tool-Specific Spans +### What Gets Traced -For more granular tracing, create spans specifically for tool executions: +The middleware automatically creates spans for: + +- **Tool Calls** (`tool.{name}`): Includes tool name, arguments, success status +- **Resource Reads** (`resource.read`): Includes resource URI +- **Prompt Retrievals** (`prompt.{name}`): Includes prompt name and arguments +- **List Operations**: Includes count of items returned + - `tools.list` + - `resources.list` + - `resource_templates.list` + - `prompts.list` + +All spans include: +- MCP method name +- Source (client/server) +- Message type (request/notification) +- Success/error status +- Exception details on failure + +### Custom Tracing Middleware + +If you need additional custom spans beyond what the built-in middleware provides, you can extend the `Middleware` base class: ```python from opentelemetry import trace from opentelemetry.trace import Status, StatusCode from fastmcp.server.middleware import Middleware, MiddlewareContext -class ToolTracingMiddleware(Middleware): - """Create detailed spans for tool executions.""" +class CustomTracingMiddleware(Middleware): + """Add custom spans for specific business logic.""" - def __init__(self, tracer_name: str = "fastmcp.tools"): - self.tracer = trace.get_tracer(tracer_name) + def __init__(self): + self.tracer = trace.get_tracer("my-custom-tracer") async def on_call_tool(self, context: MiddlewareContext, call_next): - """Create a span for each tool call with detailed attributes.""" + """Add custom spans around tool calls.""" tool_name = context.message.name + # Create a child span with custom attributes with self.tracer.start_as_current_span( - f"tool.{tool_name}", - attributes={ - "mcp.tool.name": tool_name, - "mcp.tool.arguments": str(context.message.arguments), - } + f"custom.{tool_name}", + attributes={"custom.attribute": "value"} ) as span: - try: - result = await call_next(context) + result = await call_next(context) + # Add custom business logic attributes + span.set_attribute("custom.result_type", type(result).__name__) + return result - # Add result metadata to span - span.set_attribute("mcp.tool.success", True) - if hasattr(result, 'content'): - span.set_attribute("mcp.tool.content_length", len(str(result.content))) - - span.set_status(Status(StatusCode.OK)) - return result - - except Exception as e: - span.set_attribute("mcp.tool.success", False) - span.set_attribute("mcp.tool.error", str(e)) - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - raise - -mcp.add_middleware(ToolTracingMiddleware()) -``` - -### Comprehensive Observability Middleware - -For production systems, create a middleware that handles all MCP operation types: - -```python -from opentelemetry import trace -from opentelemetry.trace import Status, StatusCode -from fastmcp.server.middleware import Middleware, MiddlewareContext - -class ComprehensiveTracingMiddleware(Middleware): - """Complete tracing for tools, resources, and prompts.""" - - def __init__(self, tracer_name: str = "fastmcp"): - self.tracer = trace.get_tracer(tracer_name) - - async def on_call_tool(self, context: MiddlewareContext, call_next): - """Trace tool executions.""" - return await self._trace_operation( - "tool.call", - {"tool.name": context.message.name}, - context, - call_next - ) - - async def on_read_resource(self, context: MiddlewareContext, call_next): - """Trace resource reads.""" - return await self._trace_operation( - "resource.read", - {"resource.uri": context.message.uri}, - context, - call_next - ) - - async def on_get_prompt(self, context: MiddlewareContext, call_next): - """Trace prompt retrievals.""" - return await self._trace_operation( - "prompt.get", - {"prompt.name": context.message.name}, - context, - call_next - ) - - async def _trace_operation( - self, - operation_name: str, - attributes: dict, - context: MiddlewareContext, - call_next - ): - """Helper to create spans with consistent attributes.""" - with self.tracer.start_as_current_span( - operation_name, - attributes={ - "mcp.method": context.method, - "mcp.source": context.source, - **attributes, - } - ) as span: - try: - result = await call_next(context) - span.set_status(Status(StatusCode.OK)) - return result - except Exception as e: - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - raise - -mcp.add_middleware(ComprehensiveTracingMiddleware()) +# Stack middleware - built-in first, then custom +mcp.add_middleware(OpenTelemetryMiddleware()) # Built-in tracing +mcp.add_middleware(CustomTracingMiddleware()) # Your custom spans ``` ## Complete Example @@ -305,8 +246,7 @@ from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogEx from fastmcp import FastMCP from fastmcp.utilities.logging import get_logger -from fastmcp.server.middleware import Middleware, MiddlewareContext -from opentelemetry.trace import Status, StatusCode +from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware # Configure OpenTelemetry resource = Resource(attributes={ @@ -325,25 +265,6 @@ logger_provider = LoggerProvider(resource=resource) logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) set_logger_provider(logger_provider) -# Middleware for tracing -class TracingMiddleware(Middleware): - def __init__(self): - self.tracer = trace.get_tracer("weather-server") - - async def on_call_tool(self, context: MiddlewareContext, call_next): - with self.tracer.start_as_current_span( - f"tool.{context.message.name}", - attributes={"tool.name": context.message.name} - ) as span: - try: - result = await call_next(context) - span.set_status(Status(StatusCode.OK)) - return result - except Exception as e: - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - raise - # Create FastMCP server mcp = FastMCP("Weather Server") @@ -351,8 +272,8 @@ mcp = FastMCP("Weather Server") logger = get_logger("weather") logger.addHandler(LoggingHandler(logger_provider=logger_provider)) -# Add tracing middleware -mcp.add_middleware(TracingMiddleware()) +# Add built-in tracing middleware +mcp.add_middleware(OpenTelemetryMiddleware()) @mcp.tool() def get_weather(city: str) -> dict: @@ -484,26 +405,26 @@ async def on_call_tool(self, context: MiddlewareContext, call_next): return await call_next(context) ``` -### Integration with FastMCP Middleware +### Integration with Other Middleware -OpenTelemetry middleware works seamlessly with FastMCP's built-in middleware: +OpenTelemetry middleware works seamlessly with FastMCP's other built-in middleware: ```python +from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware from fastmcp.server.middleware.timing import TimingMiddleware from fastmcp.server.middleware.logging import LoggingMiddleware -# Order matters: error handling first, then tracing, then logging -mcp.add_middleware(ErrorHandlingMiddleware()) -mcp.add_middleware(OpenTelemetryMiddleware()) # Your custom middleware -mcp.add_middleware(TimingMiddleware()) # Built-in timing -mcp.add_middleware(LoggingMiddleware()) # Built-in logging +# Order matters for middleware execution +mcp.add_middleware(OpenTelemetryMiddleware()) # Tracing first for complete lifecycle +mcp.add_middleware(TimingMiddleware()) # Timing within traces +mcp.add_middleware(LoggingMiddleware()) # Logging captures everything ``` The execution order ensures: -1. Errors are handled consistently -2. OpenTelemetry captures complete request lifecycle -3. Timing data is included in spans -4. Everything is logged with proper context +1. OpenTelemetry captures the complete request lifecycle including timing and logging +2. Timing data is included within trace spans +3. Logs are correlated with active traces +4. Everything is properly instrumented for observability ## Additional Resources diff --git a/examples/opentelemetry_example.py b/examples/opentelemetry_example.py index 9d21c4649..69dbe394e 100644 --- a/examples/opentelemetry_example.py +++ b/examples/opentelemetry_example.py @@ -27,10 +27,9 @@ from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogEx from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter -from opentelemetry.trace import Status, StatusCode from fastmcp import FastMCP -from fastmcp.server.middleware import Middleware, MiddlewareContext +from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware from fastmcp.utilities.logging import get_logger # ============================================================================ @@ -56,50 +55,6 @@ logger_provider = LoggerProvider(resource=resource) logger_provider.add_log_record_processor(BatchLogRecordProcessor(ConsoleLogExporter())) set_logger_provider(logger_provider) -# ============================================================================ -# Custom Middleware for OpenTelemetry Spans -# ============================================================================ - - -class OpenTelemetryMiddleware(Middleware): - """Middleware that creates OpenTelemetry spans for MCP operations.""" - - def __init__(self, tracer_name: str = "fastmcp"): - self.tracer = trace.get_tracer(tracer_name) - - async def on_call_tool(self, context: MiddlewareContext, call_next): - """Create a span for each tool call with detailed attributes.""" - tool_name = context.message.name - - # Create a span for this tool call - with self.tracer.start_as_current_span( - f"tool.{tool_name}", - attributes={ - "mcp.method": context.method, - "mcp.source": context.source, - "mcp.tool.name": tool_name, - "mcp.tool.arguments": str(context.message.arguments), - }, - ) as span: - try: - # Execute the tool - result = await call_next(context) - - # Mark span as successful - span.set_attribute("mcp.tool.success", True) - span.set_status(Status(StatusCode.OK)) - - return result - - except Exception as e: - # Record the error in the span - span.set_attribute("mcp.tool.success", False) - span.set_attribute("mcp.tool.error", str(e)) - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - raise - - # ============================================================================ # FastMCP Server Setup # ============================================================================ diff --git a/pyproject.toml b/pyproject.toml index 93b5043de..9f381b813 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,11 +47,15 @@ classifiers = [ [project.optional-dependencies] openai = ["openai>=1.102.0"] +opentelemetry = [ + "opentelemetry-api>=1.20.0", + "opentelemetry-sdk>=1.20.0", +] [dependency-groups] dev = [ "dirty-equals>=0.9.0", - "fastmcp[openai]", + "fastmcp[openai,opentelemetry]", # add optional dependencies for fastmcp dev "fastapi>=0.115.12", "inline-snapshot[dirty-equals]>=0.27.2", diff --git a/src/fastmcp/server/middleware/opentelemetry.py b/src/fastmcp/server/middleware/opentelemetry.py new file mode 100644 index 000000000..ba40f75df --- /dev/null +++ b/src/fastmcp/server/middleware/opentelemetry.py @@ -0,0 +1,313 @@ +"""OpenTelemetry instrumentation middleware for distributed tracing and observability. + +This middleware provides automatic OpenTelemetry instrumentation for FastMCP servers, +creating spans for all MCP operations. It gracefully handles the case where OpenTelemetry +is not installed, making it safe to enable by default. + +Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("MyServer") + mcp.add_middleware(OpenTelemetryMiddleware()) # Enabled by default + ``` + + To configure OpenTelemetry, set up providers before creating your server: + + ```python + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter + + # Configure tracing + trace_provider = TracerProvider() + trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) + trace.set_tracer_provider(trace_provider) + + # Now create your FastMCP server + mcp = FastMCP("MyServer") + mcp.add_middleware(OpenTelemetryMiddleware()) + ``` +""" + +import logging +from typing import Any + +from .middleware import CallNext, Middleware, MiddlewareContext + +logger = logging.getLogger(__name__) + +# Try to import OpenTelemetry components +try: + from opentelemetry import trace + from opentelemetry.trace import Status, StatusCode + + OPENTELEMETRY_AVAILABLE = True +except ImportError: + OPENTELEMETRY_AVAILABLE = False + logger.debug("OpenTelemetry not available - spans will not be created") + + +class OpenTelemetryMiddleware(Middleware): + """Middleware that creates OpenTelemetry spans for MCP operations. + + This middleware automatically instruments FastMCP servers with OpenTelemetry + distributed tracing. It creates spans for all MCP operations including tool calls, + resource reads, prompt retrievals, and list operations. + + If OpenTelemetry is not installed, this middleware becomes a no-op, making it + safe to enable by default without requiring OpenTelemetry as a dependency. + + Args: + tracer_name: Name for the OpenTelemetry tracer (default: "fastmcp") + enabled: Whether to enable tracing (default: True) + include_arguments: Whether to include operation arguments as span attributes + (default: True). Set to False to avoid including potentially sensitive data. + max_argument_length: Maximum length of argument strings in span attributes + (default: 500). Prevents spans from becoming too large. + + Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("MyServer") + + # Enable with default settings + mcp.add_middleware(OpenTelemetryMiddleware()) + + # Or customize the configuration + mcp.add_middleware(OpenTelemetryMiddleware( + tracer_name="my-custom-tracer", + include_arguments=False, # Don't include arguments for privacy + max_argument_length=1000 + )) + ``` + """ + + def __init__( + self, + tracer_name: str = "fastmcp", + enabled: bool = True, + include_arguments: bool = True, + max_argument_length: int = 500, + ): + """Initialize OpenTelemetry middleware. + + Args: + tracer_name: Name for the OpenTelemetry tracer + enabled: Whether to enable tracing + include_arguments: Whether to include operation arguments as span attributes + max_argument_length: Maximum length of argument strings in span attributes + """ + self.enabled = enabled and OPENTELEMETRY_AVAILABLE + self.include_arguments = include_arguments + self.max_argument_length = max_argument_length + + if self.enabled: + self.tracer = trace.get_tracer(tracer_name) + else: + self.tracer = None + + if not OPENTELEMETRY_AVAILABLE and enabled: + logger.info( + "OpenTelemetry middleware is enabled but opentelemetry-api is not installed. " + "Install with: pip install opentelemetry-api opentelemetry-sdk" + ) + + def _truncate_value(self, value: Any) -> str: + """Truncate a value to the configured maximum length.""" + str_value = str(value) + if len(str_value) > self.max_argument_length: + return str_value[: self.max_argument_length] + "..." + return str_value + + def _create_span_attributes(self, context: MiddlewareContext, **extra: Any) -> dict: + """Create span attributes from context and extra parameters.""" + attributes = { + "mcp.method": context.method or "unknown", + "mcp.source": context.source, + "mcp.type": context.type, + } + + if self.include_arguments: + attributes.update(extra) + + return attributes + + async def on_call_tool( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for tool execution.""" + if not self.enabled: + return await call_next(context) + + tool_name = getattr(context.message, "name", "unknown") + tool_arguments = getattr(context.message, "arguments", {}) + + span_attributes = self._create_span_attributes( + context, + **{ + "mcp.tool.name": tool_name, + "mcp.tool.arguments": self._truncate_value(tool_arguments), + }, + ) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + f"tool.{tool_name}", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_attribute("mcp.tool.success", True) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_attribute("mcp.tool.success", False) + span.set_attribute("mcp.tool.error", str(e)) + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_read_resource( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for resource reading.""" + if not self.enabled: + return await call_next(context) + + resource_uri = getattr(context.message, "uri", "unknown") + + span_attributes = self._create_span_attributes( + context, **{"mcp.resource.uri": resource_uri} + ) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + f"resource.read", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_get_prompt( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for prompt retrieval.""" + if not self.enabled: + return await call_next(context) + + prompt_name = getattr(context.message, "name", "unknown") + prompt_arguments = getattr(context.message, "arguments", {}) + + span_attributes = self._create_span_attributes( + context, + **{ + "mcp.prompt.name": prompt_name, + "mcp.prompt.arguments": self._truncate_value(prompt_arguments), + }, + ) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + f"prompt.{prompt_name}", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_list_tools( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for listing tools.""" + if not self.enabled: + return await call_next(context) + + span_attributes = self._create_span_attributes(context) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + "tools.list", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_attribute("mcp.tools.count", len(result)) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_list_resources( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for listing resources.""" + if not self.enabled: + return await call_next(context) + + span_attributes = self._create_span_attributes(context) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + "resources.list", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_attribute("mcp.resources.count", len(result)) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_list_resource_templates( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for listing resource templates.""" + if not self.enabled: + return await call_next(context) + + span_attributes = self._create_span_attributes(context) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + "resource_templates.list", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_attribute("mcp.resource_templates.count", len(result)) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise + + async def on_list_prompts( + self, context: MiddlewareContext, call_next: CallNext + ) -> Any: + """Create a span for listing prompts.""" + if not self.enabled: + return await call_next(context) + + span_attributes = self._create_span_attributes(context) + + with self.tracer.start_as_current_span( # type: ignore[union-attr] + "prompts.list", attributes=span_attributes + ) as span: + try: + result = await call_next(context) + span.set_attribute("mcp.prompts.count", len(result)) + span.set_status(Status(StatusCode.OK)) + return result + except Exception as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.record_exception(e) + raise diff --git a/tests/server/middleware/test_opentelemetry_middleware.py b/tests/server/middleware/test_opentelemetry_middleware.py new file mode 100644 index 000000000..298f059f4 --- /dev/null +++ b/tests/server/middleware/test_opentelemetry_middleware.py @@ -0,0 +1,180 @@ +"""Tests for OpenTelemetry middleware.""" + +import pytest + +from fastmcp import Client, FastMCP + + +class TestOpenTelemetryMiddlewareWithoutOTel: + """Test OpenTelemetry middleware behavior when opentelemetry is not installed.""" + + async def test_middleware_no_op_without_opentelemetry(self): + """Test that middleware works as a no-op when OpenTelemetry is not installed.""" + # Import after potentially uninstalling opentelemetry + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.tool() + def test_tool(value: str) -> str: + return f"result: {value}" + + # Should work without errors even if OpenTelemetry is not installed + async with Client(mcp) as client: + result = await client.call_tool("test_tool", {"value": "test"}) + assert result.content[0].text == "result: test" # type: ignore[attr-defined] + + +class TestOpenTelemetryMiddlewareConfiguration: + """Test OpenTelemetry middleware configuration options.""" + + def test_middleware_can_be_disabled(self): + """Test that middleware can be explicitly disabled.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + middleware = OpenTelemetryMiddleware(enabled=False) + mcp.add_middleware(middleware) + + assert middleware.enabled is False + assert middleware.tracer is None + + def test_middleware_respects_include_arguments(self): + """Test that include_arguments parameter is respected.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + middleware = OpenTelemetryMiddleware(include_arguments=False) + assert middleware.include_arguments is False + + middleware = OpenTelemetryMiddleware(include_arguments=True) + assert middleware.include_arguments is True + + def test_middleware_respects_max_argument_length(self): + """Test that max_argument_length parameter is respected.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + middleware = OpenTelemetryMiddleware(max_argument_length=100) + assert middleware.max_argument_length == 100 + + # Test truncation + long_value = "x" * 200 + truncated = middleware._truncate_value(long_value) + assert len(truncated) == 103 # 100 + "..." + assert truncated.endswith("...") + + +class TestOpenTelemetryMiddlewareOperations: + """Test that middleware handles different MCP operations.""" + + async def test_tool_call_without_errors(self): + """Test that middleware handles tool calls without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.tool() + def test_tool(value: str) -> str: + return f"result: {value}" + + async with Client(mcp) as client: + result = await client.call_tool("test_tool", {"value": "test"}) + assert result.content[0].text == "result: test" # type: ignore[attr-defined] + + async def test_resource_read_without_errors(self): + """Test that middleware handles resource reads without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.resource("test://resource") + def test_resource() -> str: + return "resource content" + + async with Client(mcp) as client: + result = await client.read_resource("test://resource") + assert result[0].text == "resource content" # type: ignore[attr-defined] + + async def test_prompt_get_without_errors(self): + """Test that middleware handles prompt retrieval without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.prompt() + def test_prompt(name: str) -> str: + return f"Hello, {name}!" + + async with Client(mcp) as client: + result = await client.get_prompt("test_prompt", {"name": "World"}) + assert any("Hello, World!" in str(msg) for msg in result.messages) + + async def test_list_tools_without_errors(self): + """Test that middleware handles list tools without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.tool() + def test_tool() -> str: + return "test" + + async with Client(mcp) as client: + result = await client.list_tools() + assert len(result) == 1 + assert result[0].name == "test_tool" + + async def test_list_resources_without_errors(self): + """Test that middleware handles list resources without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.resource("test://resource") + def test_resource() -> str: + return "test" + + async with Client(mcp) as client: + result = await client.list_resources() + assert len(result) == 1 + assert str(result[0].uri) == "test://resource" + + async def test_list_prompts_without_errors(self): + """Test that middleware handles list prompts without errors.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.prompt() + def test_prompt() -> str: + return "test" + + async with Client(mcp) as client: + result = await client.list_prompts() + assert len(result) == 1 + assert result[0].name == "test_prompt" + + +class TestOpenTelemetryMiddlewareErrorHandling: + """Test that middleware properly handles errors.""" + + async def test_tool_error_propagates(self): + """Test that errors in tools are properly propagated.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware()) + + @mcp.tool() + def failing_tool() -> str: + raise ValueError("Test error") + + async with Client(mcp) as client: + with pytest.raises(Exception): + await client.call_tool("failing_tool", {}) diff --git a/uv.lock b/uv.lock index 808099901..7c5a3a876 100644 --- a/uv.lock +++ b/uv.lock @@ -575,12 +575,16 @@ dependencies = [ openai = [ { name = "openai" }, ] +opentelemetry = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-sdk" }, +] [package.dev-dependencies] dev = [ { name = "dirty-equals" }, { name = "fastapi" }, - { name = "fastmcp", extra = ["openai"] }, + { name = "fastmcp", extra = ["openai", "opentelemetry"] }, { name = "inline-snapshot", extra = ["dirty-equals"] }, { name = "ipython", version = "8.37.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "ipython", version = "9.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, @@ -613,6 +617,8 @@ requires-dist = [ { name = "mcp", specifier = ">=1.23.1" }, { name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" }, { name = "openapi-pydantic", specifier = ">=0.5.1" }, + { name = "opentelemetry-api", marker = "extra == 'opentelemetry'", specifier = ">=1.20.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'opentelemetry'", specifier = ">=1.20.0" }, { name = "platformdirs", specifier = ">=4.0.0" }, { name = "py-key-value-aio", extras = ["disk", "memory"], specifier = ">=0.2.8,<0.4.0" }, { name = "pydantic", extras = ["email"], specifier = ">=2.11.7" }, @@ -622,13 +628,13 @@ requires-dist = [ { name = "uvicorn", specifier = ">=0.35" }, { name = "websockets", specifier = ">=15.0.1" }, ] -provides-extras = ["openai"] +provides-extras = ["openai", "opentelemetry"] [package.metadata.requires-dev] dev = [ { name = "dirty-equals", specifier = ">=0.9.0" }, { name = "fastapi", specifier = ">=0.115.12" }, - { name = "fastmcp", extras = ["openai"] }, + { name = "fastmcp", extras = ["openai", "opentelemetry"] }, { name = "inline-snapshot", extras = ["dirty-equals"], specifier = ">=0.27.2" }, { name = "ipython", specifier = ">=8.12.3" }, { name = "pdbpp", specifier = ">=0.11.7" }, @@ -705,6 +711,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/76/c6/c88e154df9c4e1a2a66ccf0005a88dfb2650c1dffb6f5ce603dfbd452ce3/idna-3.10-py3-none-any.whl", hash = "sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3", size = 70442, upload-time = "2024-09-15T18:07:37.964Z" }, ] +[[package]] +name = "importlib-metadata" +version = "8.7.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "zipp" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/66/650a33bd90f786193e4de4b3ad86ea60b53c89b669a5c7be931fac31cdb0/importlib_metadata-8.7.0.tar.gz", hash = "sha256:d13b81ad223b890aa16c5471f2ac3056cf76c5f10f82d6f9292f0b415f389000", size = 56641, upload-time = "2025-04-27T15:29:01.736Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/20/b0/36bd937216ec521246249be3bf9855081de4c5e06a0c9b4219dbeda50373/importlib_metadata-8.7.0-py3-none-any.whl", hash = "sha256:e5dd1551894c77868a30651cef00984d50e1002d06942a7101d34870c5f02afd", size = 27656, upload-time = "2025-04-27T15:29:00.214Z" }, +] + [[package]] name = "iniconfig" version = "2.1.0" @@ -1012,6 +1030,46 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/12/cf/03675d8bd8ecbf4445504d8071adab19f5f993676795708e36402ab38263/openapi_pydantic-0.5.1-py3-none-any.whl", hash = "sha256:a3a09ef4586f5bd760a8df7f43028b60cafb6d9f61de2acba9574766255ab146", size = 96381, upload-time = "2025-01-08T19:29:25.275Z" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.39.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "importlib-metadata" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/0b/e5428c009d4d9af0515b0a8371a8aaae695371af291f45e702f7969dce6b/opentelemetry_api-1.39.0.tar.gz", hash = "sha256:6130644268c5ac6bdffaf660ce878f10906b3e789f7e2daa5e169b047a2933b9", size = 65763, upload-time = "2025-12-03T13:19:56.378Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/05/85/d831a9bc0a9e0e1a304ff3d12c1489a5fbc9bf6690a15dcbdae372bbca45/opentelemetry_api-1.39.0-py3-none-any.whl", hash = "sha256:3c3b3ca5c5687b1b5b37e5c5027ff68eacea8675241b29f13110a8ffbb8f0459", size = 66357, upload-time = "2025-12-03T13:19:33.043Z" }, +] + +[[package]] +name = "opentelemetry-sdk" +version = "1.39.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/51/e3/7cd989003e7cde72e0becfe830abff0df55c69d237ee7961a541e0167833/opentelemetry_sdk-1.39.0.tar.gz", hash = "sha256:c22204f12a0529e07aa4d985f1bca9d6b0e7b29fe7f03e923548ae52e0e15dde", size = 171322, upload-time = "2025-12-03T13:20:09.651Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a4/b4/2adc8bc83eb1055ecb592708efb6f0c520cc2eb68970b02b0f6ecda149cf/opentelemetry_sdk-1.39.0-py3-none-any.whl", hash = "sha256:90cfb07600dfc0d2de26120cebc0c8f27e69bf77cd80ef96645232372709a514", size = 132413, upload-time = "2025-12-03T13:19:51.364Z" }, +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.60b0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/71/0e/176a7844fe4e3cb5de604212094dffaed4e18b32f1c56b5258bcbcba85c2/opentelemetry_semantic_conventions-0.60b0.tar.gz", hash = "sha256:227d7aa73cbb8a2e418029d6b6465553aa01cf7e78ec9d0bc3255c7b3ac5bf8f", size = 137935, upload-time = "2025-12-03T13:20:12.395Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d0/56/af0306666f91bae47db14d620775604688361f0f76a872e0005277311131/opentelemetry_semantic_conventions-0.60b0-py3-none-any.whl", hash = "sha256:069530852691136018087b52688857d97bba61cd641d0f8628d2d92788c4f78a", size = 219981, upload-time = "2025-12-03T13:19:53.585Z" }, +] + [[package]] name = "packaging" version = "25.0" @@ -2103,3 +2161,12 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/68/a1/dcb68430b1d00b698ae7a7e0194433bce4f07ded185f0ee5fb21e2a2e91e/websockets-15.0.1-pp310-pypy310_pp73-win_amd64.whl", hash = "sha256:cad21560da69f4ce7658ca2cb83138fb4cf695a2ba3e475e0559e05991aa8122", size = 176884, upload-time = "2025-03-05T20:03:27.934Z" }, { url = "https://files.pythonhosted.org/packages/fa/a8/5b41e0da817d64113292ab1f8247140aac61cbf6cfd085d6a0fa77f4984f/websockets-15.0.1-py3-none-any.whl", hash = "sha256:f7a866fbc1e97b5c617ee4116daaa09b722101d4a3c170c787450ba409f9736f", size = 169743, upload-time = "2025-03-05T20:03:39.41Z" }, ] + +[[package]] +name = "zipp" +version = "3.23.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e3/02/0f2892c661036d50ede074e376733dca2ae7c6eb617489437771209d4180/zipp-3.23.0.tar.gz", hash = "sha256:a07157588a12518c9d4034df3fbbee09c814741a33ff63c05fa29d26a2404166", size = 25547, upload-time = "2025-06-08T17:06:39.4Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2e/54/647ade08bf0db230bfea292f893923872fd20be6ac6f53b2b936ba839d75/zipp-3.23.0-py3-none-any.whl", hash = "sha256:071652d6115ed432f5ce1d34c336c0adfd6a884660d1e9712a256d3d3bd4b14e", size = 10276, upload-time = "2025-06-08T17:06:38.034Z" }, +] From a9b6e7196615695304916d586cbd00d077e39485 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Wed, 3 Dec 2025 23:58:17 +0000 Subject: [PATCH 3/3] Add trace context propagation to OpenTelemetry middleware Enable distributed tracing across protocols that don't support HTTP headers (like SSE) by propagating W3C Trace Context through MCP _meta fields. - Add propagate_context parameter (default: True) to OpenTelemetryMiddleware - Implement _extract_trace_context() to read traceparent/tracestate from request metadata - Implement _inject_trace_context() to write trace context to response metadata - Update all operation handlers to extract parent context and inject into results - Add comprehensive test coverage for context propagation - Update documentation with examples and configuration details This enables trace continuity across MCP calls, allowing clients to link server spans to their traces and propagate context downstream. Co-authored-by: William Easton --- docs/integrations/opentelemetry.mdx | 37 ++++- .../server/middleware/opentelemetry.py | 144 ++++++++++++++++-- .../test_opentelemetry_middleware.py | 125 +++++++++++++++ 3 files changed, 294 insertions(+), 12 deletions(-) diff --git a/docs/integrations/opentelemetry.mdx b/docs/integrations/opentelemetry.mdx index 6a27e1dea..7dced66f5 100644 --- a/docs/integrations/opentelemetry.mdx +++ b/docs/integrations/opentelemetry.mdx @@ -173,10 +173,45 @@ mcp.add_middleware(OpenTelemetryMiddleware( tracer_name="my-custom-tracer", # Custom tracer name enabled=True, # Explicitly enable/disable include_arguments=False, # Don't include arguments for privacy - max_argument_length=1000 # Limit argument string length in spans + max_argument_length=1000, # Limit argument string length in spans + propagate_context=True # Enable trace context propagation (default) )) ``` +### Trace Context Propagation + +The middleware automatically propagates trace context across MCP calls using the `_meta` field. This enables distributed tracing even when using protocols like SSE that don't support standard HTTP headers. + +**How it works:** + +1. **Incoming requests**: The middleware extracts trace context (W3C `traceparent` and `tracestate`) from the request's `_meta` field +2. **Span creation**: New spans are created as children of the incoming trace context +3. **Outgoing responses**: The middleware injects the current trace context into the response's `_meta` field + +This means that if a client includes trace context in their request metadata, your server's spans will be linked to the client's trace. Similarly, if your server calls another MCP server, you can propagate context downstream. + +**Example: Client propagating context** + +```python +# Client code - sending trace context +result = await client.call_tool( + "my_tool", + {"arg": "value"}, + _meta={"traceparent": "00-trace_id-span_id-01", "tracestate": "vendor=value"} +) + +# The server will create spans as children of this trace +# And the response will include updated trace context in result.meta +if result.meta: + downstream_traceparent = result.meta.get("traceparent") +``` + +To disable context propagation: + +```python +mcp.add_middleware(OpenTelemetryMiddleware(propagate_context=False)) +``` + ### What Gets Traced The middleware automatically creates spans for: diff --git a/src/fastmcp/server/middleware/opentelemetry.py b/src/fastmcp/server/middleware/opentelemetry.py index ba40f75df..8b02573f3 100644 --- a/src/fastmcp/server/middleware/opentelemetry.py +++ b/src/fastmcp/server/middleware/opentelemetry.py @@ -41,7 +41,11 @@ logger = logging.getLogger(__name__) # Try to import OpenTelemetry components try: from opentelemetry import trace + from opentelemetry.context import Context from opentelemetry.trace import Status, StatusCode + from opentelemetry.trace.propagation.tracecontext import ( + TraceContextTextMapPropagator, + ) OPENTELEMETRY_AVAILABLE = True except ImportError: @@ -66,6 +70,10 @@ class OpenTelemetryMiddleware(Middleware): (default: True). Set to False to avoid including potentially sensitive data. max_argument_length: Maximum length of argument strings in span attributes (default: 500). Prevents spans from becoming too large. + propagate_context: Whether to propagate trace context through MCP _meta fields + (default: True). When enabled, trace context is injected into response metadata + and extracted from request metadata, enabling distributed tracing across protocols + that don't support HTTP headers (like SSE). Example: ```python @@ -81,7 +89,8 @@ class OpenTelemetryMiddleware(Middleware): mcp.add_middleware(OpenTelemetryMiddleware( tracer_name="my-custom-tracer", include_arguments=False, # Don't include arguments for privacy - max_argument_length=1000 + max_argument_length=1000, + propagate_context=True # Enable trace context propagation )) ``` """ @@ -92,6 +101,7 @@ class OpenTelemetryMiddleware(Middleware): enabled: bool = True, include_arguments: bool = True, max_argument_length: int = 500, + propagate_context: bool = True, ): """Initialize OpenTelemetry middleware. @@ -100,15 +110,22 @@ class OpenTelemetryMiddleware(Middleware): enabled: Whether to enable tracing include_arguments: Whether to include operation arguments as span attributes max_argument_length: Maximum length of argument strings in span attributes + propagate_context: Whether to propagate trace context through MCP _meta fields """ self.enabled = enabled and OPENTELEMETRY_AVAILABLE self.include_arguments = include_arguments self.max_argument_length = max_argument_length + self.propagate_context = propagate_context if self.enabled: self.tracer = trace.get_tracer(tracer_name) + if self.propagate_context: + self.propagator = TraceContextTextMapPropagator() + else: + self.propagator = None else: self.tracer = None + self.propagator = None if not OPENTELEMETRY_AVAILABLE and enabled: logger.info( @@ -123,6 +140,85 @@ class OpenTelemetryMiddleware(Middleware): return str_value[: self.max_argument_length] + "..." return str_value + def _extract_trace_context(self, context: MiddlewareContext) -> Context | None: + """Extract trace context from request metadata if available. + + Args: + context: The middleware context containing the request + + Returns: + OpenTelemetry Context with extracted trace information, or None if not available + """ + if not self.propagate_context or not self.propagator: + return None + + # Get _meta from the request message + request_meta = getattr(context.message, "_meta", None) + if not request_meta or not isinstance(request_meta, dict): + return None + + # Extract trace context using W3C Trace Context format + try: + carrier = {} + if "traceparent" in request_meta: + carrier["traceparent"] = request_meta["traceparent"] + if "tracestate" in request_meta: + carrier["tracestate"] = request_meta["tracestate"] + + if carrier: + otel_context = self.propagator.extract(carrier=carrier) # type: ignore[union-attr] + return otel_context + except Exception as e: + logger.debug(f"Failed to extract trace context from metadata: {e}") + + return None + + def _inject_trace_context(self, result: Any) -> Any: + """Inject current trace context into result metadata. + + Args: + result: The result to inject trace context into + + Returns: + Result with trace context injected into _meta field + """ + if not self.propagate_context or not self.propagator: + return result + + try: + # Get current span context + current_span = trace.get_current_span() + if not current_span or not current_span.get_span_context().is_valid: + return result + + # Inject trace context into carrier + carrier: dict[str, str] = {} + self.propagator.inject(carrier=carrier) # type: ignore[union-attr] + + if not carrier: + return result + + # Add trace context to result metadata + # Handle different result types + if hasattr(result, "_meta"): + # Result already has _meta attribute (like CallToolResult) + if result._meta is None: + result._meta = {} + result._meta.update(carrier) + elif hasattr(result, "meta"): + # Result has meta attribute (like ToolResult) + if result.meta is None: + result.meta = {} + result.meta.update(carrier) + else: + # For list results or other types, we can't inject context + pass + + except Exception as e: + logger.debug(f"Failed to inject trace context into metadata: {e}") + + return result + def _create_span_attributes(self, context: MiddlewareContext, **extra: Any) -> dict: """Create span attributes from context and extra parameters.""" attributes = { @@ -143,6 +239,9 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + tool_name = getattr(context.message, "name", "unknown") tool_arguments = getattr(context.message, "arguments", {}) @@ -154,14 +253,16 @@ class OpenTelemetryMiddleware(Middleware): }, ) + # Start span with parent context if available with self.tracer.start_as_current_span( # type: ignore[union-attr] - f"tool.{tool_name}", attributes=span_attributes + f"tool.{tool_name}", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) span.set_attribute("mcp.tool.success", True) span.set_status(Status(StatusCode.OK)) - return result + # Inject trace context into result + return self._inject_trace_context(result) except Exception as e: span.set_attribute("mcp.tool.success", False) span.set_attribute("mcp.tool.error", str(e)) @@ -176,6 +277,9 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + resource_uri = getattr(context.message, "uri", "unknown") span_attributes = self._create_span_attributes( @@ -183,12 +287,12 @@ class OpenTelemetryMiddleware(Middleware): ) with self.tracer.start_as_current_span( # type: ignore[union-attr] - f"resource.read", attributes=span_attributes + "resource.read", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) span.set_status(Status(StatusCode.OK)) - return result + return self._inject_trace_context(result) except Exception as e: span.set_status(Status(StatusCode.ERROR, str(e))) span.record_exception(e) @@ -201,6 +305,9 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + prompt_name = getattr(context.message, "name", "unknown") prompt_arguments = getattr(context.message, "arguments", {}) @@ -213,12 +320,12 @@ class OpenTelemetryMiddleware(Middleware): ) with self.tracer.start_as_current_span( # type: ignore[union-attr] - f"prompt.{prompt_name}", attributes=span_attributes + f"prompt.{prompt_name}", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) span.set_status(Status(StatusCode.OK)) - return result + return self._inject_trace_context(result) except Exception as e: span.set_status(Status(StatusCode.ERROR, str(e))) span.record_exception(e) @@ -231,15 +338,19 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + span_attributes = self._create_span_attributes(context) with self.tracer.start_as_current_span( # type: ignore[union-attr] - "tools.list", attributes=span_attributes + "tools.list", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) span.set_attribute("mcp.tools.count", len(result)) span.set_status(Status(StatusCode.OK)) + # List operations return lists, so we can't inject trace context return result except Exception as e: span.set_status(Status(StatusCode.ERROR, str(e))) @@ -253,10 +364,13 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + span_attributes = self._create_span_attributes(context) with self.tracer.start_as_current_span( # type: ignore[union-attr] - "resources.list", attributes=span_attributes + "resources.list", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) @@ -275,10 +389,15 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + span_attributes = self._create_span_attributes(context) with self.tracer.start_as_current_span( # type: ignore[union-attr] - "resource_templates.list", attributes=span_attributes + "resource_templates.list", + attributes=span_attributes, + context=parent_context, ) as span: try: result = await call_next(context) @@ -297,10 +416,13 @@ class OpenTelemetryMiddleware(Middleware): if not self.enabled: return await call_next(context) + # Extract trace context from request metadata + parent_context = self._extract_trace_context(context) + span_attributes = self._create_span_attributes(context) with self.tracer.start_as_current_span( # type: ignore[union-attr] - "prompts.list", attributes=span_attributes + "prompts.list", attributes=span_attributes, context=parent_context ) as span: try: result = await call_next(context) diff --git a/tests/server/middleware/test_opentelemetry_middleware.py b/tests/server/middleware/test_opentelemetry_middleware.py index 298f059f4..5fd821415 100644 --- a/tests/server/middleware/test_opentelemetry_middleware.py +++ b/tests/server/middleware/test_opentelemetry_middleware.py @@ -178,3 +178,128 @@ class TestOpenTelemetryMiddlewareErrorHandling: async with Client(mcp) as client: with pytest.raises(Exception): await client.call_tool("failing_tool", {}) + + +class TestOpenTelemetryMiddlewareContextPropagation: + """Test trace context propagation through MCP _meta fields.""" + + def test_propagate_context_can_be_disabled(self): + """Test that context propagation can be disabled.""" + from fastmcp.server.middleware.opentelemetry import OpenTelemetryMiddleware + + middleware = OpenTelemetryMiddleware(propagate_context=False) + assert middleware.propagate_context is False + assert middleware.propagator is None + + def test_propagate_context_enabled_by_default(self): + """Test that context propagation is enabled by default when OTel is available.""" + from fastmcp.server.middleware.opentelemetry import ( + OPENTELEMETRY_AVAILABLE, + OpenTelemetryMiddleware, + ) + + middleware = OpenTelemetryMiddleware() + assert middleware.propagate_context is True + if OPENTELEMETRY_AVAILABLE: + assert middleware.propagator is not None + else: + assert middleware.propagator is None + + async def test_trace_context_injection_in_tool_result(self): + """Test that trace context is injected into tool result metadata.""" + from fastmcp.server.middleware.opentelemetry import ( + OPENTELEMETRY_AVAILABLE, + OpenTelemetryMiddleware, + ) + + if not OPENTELEMETRY_AVAILABLE: + pytest.skip("OpenTelemetry not available") + + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + + # Set up a tracer provider + trace.set_tracer_provider(TracerProvider()) + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware(propagate_context=True)) + + @mcp.tool() + def test_tool(value: str) -> str: + return f"result: {value}" + + async with Client(mcp) as client: + result = await client.call_tool("test_tool", {"value": "test"}) + # Check that result has metadata with trace context + assert result.meta is not None + assert "traceparent" in result.meta + + async def test_trace_context_extraction_from_request(self): + """Test that trace context is extracted from request metadata.""" + from fastmcp.server.middleware.opentelemetry import ( + OPENTELEMETRY_AVAILABLE, + OpenTelemetryMiddleware, + ) + + if not OPENTELEMETRY_AVAILABLE: + pytest.skip("OpenTelemetry not available") + + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + + # Set up a tracer provider + trace_provider = TracerProvider() + trace.set_tracer_provider(trace_provider) + + mcp = FastMCP("Test") + middleware = OpenTelemetryMiddleware(propagate_context=True) + mcp.add_middleware(middleware) + + # Track whether context was extracted + extracted_context = [] + + original_extract = middleware._extract_trace_context + + def mock_extract(context): + result = original_extract(context) + extracted_context.append(result) + return result + + middleware._extract_trace_context = mock_extract # type: ignore[method-assign] + + @mcp.tool() + def test_tool(value: str) -> str: + return f"result: {value}" + + async with Client(mcp) as client: + # Call tool without trace context - should extract None + await client.call_tool("test_tool", {"value": "test"}) + assert len(extracted_context) > 0 + + async def test_context_propagation_disabled_no_injection(self): + """Test that no context is injected when propagation is disabled.""" + from fastmcp.server.middleware.opentelemetry import ( + OPENTELEMETRY_AVAILABLE, + OpenTelemetryMiddleware, + ) + + if not OPENTELEMETRY_AVAILABLE: + pytest.skip("OpenTelemetry not available") + + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + + # Set up a tracer provider + trace.set_tracer_provider(TracerProvider()) + + mcp = FastMCP("Test") + mcp.add_middleware(OpenTelemetryMiddleware(propagate_context=False)) + + @mcp.tool() + def test_tool(value: str) -> str: + return f"result: {value}" + + async with Client(mcp) as client: + result = await client.call_tool("test_tool", {"value": "test"}) + # Check that result has no trace context metadata + assert result.meta is None or "traceparent" not in result.meta