mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
046f845ddf
Adds opt-in distributed tracing via OpenTelemetry for observability into
FastMCP server and client operations.
Server spans are created for tool calls, resource reads, and prompt
renders with attributes like component key, component type, provider
type, session ID, and auth context. Client spans wrap outgoing calls
with trace context propagation via W3C headers in request meta.
Components provide their own span attributes through a `get_span_attributes()`
method that subclasses override - this lets LocalProvider, FastMCPProvider,
and ProxyProvider each include relevant context (original names, backend URIs).
To enable: configure an OpenTelemetry SDK with a TracerProvider before
importing fastmcp. Traces export to any OTLP-compatible backend.
Closes ENG-2813
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import asyncio
|
|
import socket
|
|
import sys
|
|
from collections.abc import Callable, Generator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from opentelemetry import trace
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
|
|
|
from fastmcp.utilities.tests import temporary_settings
|
|
|
|
# Use SelectorEventLoop on Windows to avoid ProactorEventLoop crashes
|
|
# See: https://github.com/python/cpython/issues/116773
|
|
if sys.platform == "win32":
|
|
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
|
|
|
|
|
def pytest_collection_modifyitems(items):
|
|
"""Automatically mark tests in integration_tests folder with 'integration' marker."""
|
|
for item in items:
|
|
# Check if the test is in the integration_tests folder
|
|
if "integration_tests" in str(item.fspath):
|
|
item.add_marker(pytest.mark.integration)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def import_rich_rule():
|
|
# What a hack
|
|
import rich.rule # noqa: F401
|
|
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolate_settings_home(tmp_path: Path):
|
|
"""Ensure each test uses an isolated settings.home directory.
|
|
|
|
This prevents SQLite database locking issues on Windows when multiple
|
|
tests share the same DiskStore directory in settings.home / "oauth-proxy".
|
|
"""
|
|
test_home = tmp_path / "fastmcp-test-home"
|
|
test_home.mkdir(exist_ok=True)
|
|
|
|
with temporary_settings(home=test_home):
|
|
yield
|
|
|
|
|
|
def get_fn_name(fn: Callable[..., Any]) -> str:
|
|
return fn.__name__ # ty: ignore[unresolved-attribute]
|
|
|
|
|
|
@pytest.fixture
|
|
def worker_id(request):
|
|
"""Get the xdist worker ID, or 'master' if not using xdist."""
|
|
return getattr(request.config, "workerinput", {}).get("workerid", "master")
|
|
|
|
|
|
@pytest.fixture
|
|
def free_port():
|
|
"""Get a free port for the test to use."""
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
s.listen(1)
|
|
port = s.getsockname()[1]
|
|
return port
|
|
|
|
|
|
@pytest.fixture
|
|
def free_port_factory(worker_id):
|
|
"""Factory to get free ports that tracks used ports per test session."""
|
|
used_ports = set()
|
|
|
|
def get_port():
|
|
while True:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
s.listen(1)
|
|
port = s.getsockname()[1]
|
|
if port not in used_ports:
|
|
used_ports.add(port)
|
|
return port
|
|
|
|
return get_port
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def otel_trace_provider() -> Generator[
|
|
tuple[TracerProvider, InMemorySpanExporter], None, None
|
|
]:
|
|
"""Configure OTEL SDK with in-memory span exporter for testing.
|
|
|
|
Session-scoped because TracerProvider can only be set once per process.
|
|
"""
|
|
exporter = InMemorySpanExporter()
|
|
provider = TracerProvider()
|
|
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
trace.set_tracer_provider(provider)
|
|
yield provider, exporter
|
|
|
|
|
|
@pytest.fixture
|
|
def trace_exporter(
|
|
otel_trace_provider: tuple[TracerProvider, InMemorySpanExporter],
|
|
) -> Generator[InMemorySpanExporter, None, None]:
|
|
"""Get the span exporter and clear it between tests."""
|
|
_, exporter = otel_trace_provider
|
|
exporter.clear()
|
|
yield exporter
|
|
exporter.clear()
|