mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 22:14:18 +02:00
21db5c6cc0- Add server_span wrappers to list_tools, list_resources, list_resource_templates, and list_prompts in FastMCP server - Add client_span wrappers to list_tools_mcp, list_resources_mcp, list_resource_templates_mcp, and list_prompts_mcp in client mixins - Add optional `method` parameter to delegate_span in telemetry.py and set mcp.method.name on delegate spans - Update all delegate_span callers in fastmcp_provider.py to pass the MCP method name (tools/call, resources/read, prompts/get) - Add tests for new server list spans, client list spans, and delegate span method attributes Refs: #3887 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
"""Tests for server-level OpenTelemetry tracing on list operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
|
from opentelemetry.trace import SpanKind
|
|
|
|
from fastmcp import FastMCP
|
|
|
|
|
|
class TestListToolsTracing:
|
|
async def test_list_tools_creates_span(self, trace_exporter: InMemorySpanExporter):
|
|
mcp = FastMCP("test-server")
|
|
|
|
@mcp.tool()
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
tools = await mcp.list_tools()
|
|
assert len(tools) == 1
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "tools/list"]
|
|
assert len(list_spans) >= 1
|
|
|
|
span = list_spans[0]
|
|
assert span.kind == SpanKind.SERVER
|
|
assert span.attributes is not None
|
|
assert span.attributes["mcp.method.name"] == "tools/list"
|
|
assert span.attributes["fastmcp.server.name"] == "test-server"
|
|
assert span.attributes["fastmcp.component.type"] == "tool"
|
|
|
|
async def test_list_tools_empty_creates_span(
|
|
self, trace_exporter: InMemorySpanExporter
|
|
):
|
|
mcp = FastMCP("test-server")
|
|
|
|
tools = await mcp.list_tools()
|
|
assert len(tools) == 0
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "tools/list"]
|
|
assert len(list_spans) >= 1
|
|
|
|
|
|
class TestListResourcesTracing:
|
|
async def test_list_resources_creates_span(
|
|
self, trace_exporter: InMemorySpanExporter
|
|
):
|
|
mcp = FastMCP("test-server")
|
|
|
|
@mcp.resource("config://app")
|
|
def get_config() -> str:
|
|
return "config"
|
|
|
|
resources = await mcp.list_resources()
|
|
assert len(resources) >= 1
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "resources/list"]
|
|
assert len(list_spans) >= 1
|
|
|
|
span = list_spans[0]
|
|
assert span.kind == SpanKind.SERVER
|
|
assert span.attributes is not None
|
|
assert span.attributes["mcp.method.name"] == "resources/list"
|
|
assert span.attributes["fastmcp.server.name"] == "test-server"
|
|
assert span.attributes["fastmcp.component.type"] == "resource"
|
|
|
|
|
|
class TestListResourceTemplatesTracing:
|
|
async def test_list_resource_templates_creates_span(
|
|
self, trace_exporter: InMemorySpanExporter
|
|
):
|
|
mcp = FastMCP("test-server")
|
|
|
|
@mcp.resource("users://{user_id}/profile")
|
|
def get_profile(user_id: str) -> str:
|
|
return f"profile {user_id}"
|
|
|
|
templates = await mcp.list_resource_templates()
|
|
assert len(templates) >= 1
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "resources/templates/list"]
|
|
assert len(list_spans) >= 1
|
|
|
|
span = list_spans[0]
|
|
assert span.kind == SpanKind.SERVER
|
|
assert span.attributes is not None
|
|
assert span.attributes["mcp.method.name"] == "resources/templates/list"
|
|
assert span.attributes["fastmcp.server.name"] == "test-server"
|
|
assert span.attributes["fastmcp.component.type"] == "resource_template"
|
|
|
|
|
|
class TestListPromptsTracing:
|
|
async def test_list_prompts_creates_span(
|
|
self, trace_exporter: InMemorySpanExporter
|
|
):
|
|
mcp = FastMCP("test-server")
|
|
|
|
@mcp.prompt()
|
|
def greeting(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
prompts = await mcp.list_prompts()
|
|
assert len(prompts) == 1
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "prompts/list"]
|
|
assert len(list_spans) >= 1
|
|
|
|
span = list_spans[0]
|
|
assert span.kind == SpanKind.SERVER
|
|
assert span.attributes is not None
|
|
assert span.attributes["mcp.method.name"] == "prompts/list"
|
|
assert span.attributes["fastmcp.server.name"] == "test-server"
|
|
assert span.attributes["fastmcp.component.type"] == "prompt"
|
|
|
|
async def test_list_prompts_empty_creates_span(
|
|
self, trace_exporter: InMemorySpanExporter
|
|
):
|
|
mcp = FastMCP("test-server")
|
|
|
|
prompts = await mcp.list_prompts()
|
|
assert len(prompts) == 0
|
|
|
|
spans = trace_exporter.get_finished_spans()
|
|
list_spans = [s for s in spans if s.name == "prompts/list"]
|
|
assert len(list_spans) >= 1
|