ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix error.type compliance, remove duplicate client span, add error handling to delegate_span

Browse source 
- Use "tool_error" for ToolError exceptions and type(e).__name__ for others
  in server_span, client_span, and delegate_span (OTel spec compliance)
- Remove duplicate client_span from call_tool(); check result.isError inside
  call_tool_mcp()'s existing span instead
- Add error.type attribute and status description to delegate_span
- Fix indentation in server.py list operations (linter auto-fix)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
strawgate 2026-04-13 00:40:49 -05:00
commit 27cc3f4a8f
6 changed files with 157 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

42
src/fastmcp/client/mixins/tools.py
View file

@ -7,6 +7,7 @@ import weakref
from typing import TYPE_CHECKING, Any, Literal, overload
import mcp.types
from opentelemetry.trace import Status, StatusCode
from pydantic import RootModel
if TYPE_CHECKING:
@ -151,7 +152,7 @@ class ClientToolsMixin:
name,
session_id=self.transport.get_session_id(),
tool_name=name,
):
) as span:
logger.debug(f"[{self.name}] called call_tool: {name}")
# Inject trace context into meta for propagation to server
@ -166,6 +167,18 @@ class ClientToolsMixin:
meta=propagated_meta if propagated_meta else None,
)
)
# Reflect tool-level errors on the span so callers see ERROR
# status even though the MCP protocol call itself succeeded.
if result.isError:
span.set_attribute("error.type", "tool_error")
description = ""
if result.content and isinstance(
result.content[0], mcp.types.TextContent
):
description = result.content[0].text
span.set_status(Status(StatusCode.ERROR, description))
return result
async def _parse_call_tool_result(
@ -284,23 +297,16 @@ class ClientToolsMixin:
name, arguments, task_id, ttl, meta=request_meta or None
)
with client_span(
f"tools/call {name}",
"tools/call",
name,
session_id=self.transport.get_session_id(),
tool_name=name,
):
result = await self.call_tool_mcp(
name=name,
arguments=arguments or {},
timeout=timeout,
progress_handler=progress_handler,
meta=request_meta or None,
)
return await self._parse_call_tool_result(
name, result, raise_on_error=raise_on_error
)
result = await self.call_tool_mcp(
name=name,
arguments=arguments or {},
timeout=timeout,
progress_handler=progress_handler,
meta=request_meta or None,
)
return await self._parse_call_tool_result(
name, result, raise_on_error=raise_on_error
)
async def _call_tool_as_task(
self: Client,

7
src/fastmcp/client/telemetry.py
View file

@ -42,7 +42,12 @@ def client_span(
try:
yield span
except Exception as e:
span.set_attribute("error.type", type(e).__name__)
span.set_attribute(
"error.type",
"tool_error"
if type(e).__name__ == "ToolError"
else type(e).__name__,
)
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
raise

192
src/fastmcp/server/server.py
View file

@ -606,10 +606,7 @@ class FastMCP(
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
with server_span(
"tools/list", "tools/list", self.name, "tool", ""
):
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message=mcp.types.ListToolsRequest(method="tools/list"),
@ -623,31 +620,35 @@ class FastMCP(
call_next=lambda context: self.list_tools(run_middleware=False),
)
# Get all tools, apply session transforms, then filter enabled
# and model-visible (app-only tools are hidden from the model).
tools = list(await super().list_tools())
tools = await apply_session_transforms(tools)
tools = [t for t in tools if is_enabled(t) and _is_model_visible(t)]
# Core logic: list tools
with server_span(
"tools/list", "tools/list", self.name, "tool", ""
):
# Get all tools, apply session transforms, then filter enabled
# and model-visible (app-only tools are hidden from the model).
tools = list(await super().list_tools())
tools = await apply_session_transforms(tools)
tools = [t for t in tools if is_enabled(t) and _is_model_visible(t)]
# Rewrite per-tool Prefab renderer URIs based on the tool's
# mount-point address. The walk pairs each tool with the
# provider that yielded it, computes the hashed URI, and
# produces a model_copy with the URI in place. Original
# Tool objects are not mutated.
tools = self._rewrite_prefab_uris(tools)
# Rewrite per-tool Prefab renderer URIs based on the tool's
# mount-point address. The walk pairs each tool with the
# provider that yielded it, computes the hashed URI, and
# produces a model_copy with the URI in place. Original
# Tool objects are not mutated.
tools = self._rewrite_prefab_uris(tools)
skip_auth, token = _get_auth_context()
authorized: list[Tool] = []
for tool in tools:
if not skip_auth and tool.auth is not None:
ctx = AuthContext(token=token, component=tool)
try:
if not await run_auth_checks(tool.auth, ctx):
skip_auth, token = _get_auth_context()
authorized: list[Tool] = []
for tool in tools:
if not skip_auth and tool.auth is not None:
ctx = AuthContext(token=token, component=tool)
try:
if not await run_auth_checks(tool.auth, ctx):
continue
except AuthorizationError:
continue
except AuthorizationError:
continue
authorized.append(tool)
return authorized
authorized.append(tool)
return authorized
async def _get_tool(
self, name: str, version: VersionSpec | None = None
@ -743,10 +744,7 @@ class FastMCP(
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
with server_span(
"resources/list", "resources/list", self.name, "resource", ""
):
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
@ -760,32 +758,36 @@ class FastMCP(
call_next=lambda context: self.list_resources(run_middleware=False),
)
# Get all resources, apply session transforms, then filter enabled
resources = list(await super().list_resources())
resources = await apply_session_transforms(resources)
resources = [r for r in resources if is_enabled(r)]
# Core logic: list resources
with server_span(
"resources/list", "resources/list", self.name, "resource", ""
):
# Get all resources, apply session transforms, then filter enabled
resources = list(await super().list_resources())
resources = await apply_session_transforms(resources)
resources = [r for r in resources if is_enabled(r)]
# Append synthetic Prefab renderer resources — one per
# prefab tool, hashed by mount address. These don't live on
# any provider's storage; they're computed on demand.
from fastmcp.server.providers.prefab_synthesis import (
synthesize_prefab_resources,
)
# Append synthetic Prefab renderer resources — one per
# prefab tool, hashed by mount address. These don't live on
# any provider's storage; they're computed on demand.
from fastmcp.server.providers.prefab_synthesis import (
synthesize_prefab_resources,
)
resources.extend(await synthesize_prefab_resources(self))
resources.extend(await synthesize_prefab_resources(self))
skip_auth, token = _get_auth_context()
authorized: list[Resource] = []
for resource in resources:
if not skip_auth and resource.auth is not None:
ctx = AuthContext(token=token, component=resource)
try:
if not await run_auth_checks(resource.auth, ctx):
skip_auth, token = _get_auth_context()
authorized: list[Resource] = []
for resource in resources:
if not skip_auth and resource.auth is not None:
ctx = AuthContext(token=token, component=resource)
try:
if not await run_auth_checks(resource.auth, ctx):
continue
except AuthorizationError:
continue
except AuthorizationError:
continue
authorized.append(resource)
return authorized
authorized.append(resource)
return authorized
async def _get_resource(
self, uri: str, version: VersionSpec | None = None
@ -877,14 +879,7 @@ class FastMCP(
auth filtering, and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
with server_span(
"resources/templates/list",
"resources/templates/list",
self.name,
"resource_template",
"",
):
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
@ -900,23 +895,31 @@ class FastMCP(
),
)
# Get all templates, apply session transforms, then filter enabled
templates = list(await super().list_resource_templates())
templates = await apply_session_transforms(templates)
templates = [t for t in templates if is_enabled(t)]
# Core logic: list resource templates
with server_span(
"resources/templates/list",
"resources/templates/list",
self.name,
"resource_template",
"",
):
# Get all templates, apply session transforms, then filter enabled
templates = list(await super().list_resource_templates())
templates = await apply_session_transforms(templates)
templates = [t for t in templates if is_enabled(t)]
skip_auth, token = _get_auth_context()
authorized: list[ResourceTemplate] = []
for template in templates:
if not skip_auth and template.auth is not None:
ctx = AuthContext(token=token, component=template)
try:
if not await run_auth_checks(template.auth, ctx):
skip_auth, token = _get_auth_context()
authorized: list[ResourceTemplate] = []
for template in templates:
if not skip_auth and template.auth is not None:
ctx = AuthContext(token=token, component=template)
try:
if not await run_auth_checks(template.auth, ctx):
continue
except AuthorizationError:
continue
except AuthorizationError:
continue
authorized.append(template)
return authorized
authorized.append(template)
return authorized
async def _get_resource_template(
self, uri: str, version: VersionSpec | None = None
@ -1010,10 +1013,7 @@ class FastMCP(
and middleware execution. Returns all versions (no deduplication).
Protocol handlers deduplicate for MCP wire format.
"""
with server_span(
"prompts/list", "prompts/list", self.name, "prompt", ""
):
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
async with fastmcp.server.context.Context(fastmcp=self) as ctx:
if run_middleware:
mw_context = MiddlewareContext(
message={},
@ -1027,23 +1027,27 @@ class FastMCP(
call_next=lambda context: self.list_prompts(run_middleware=False),
)
# Get all prompts, apply session transforms, then filter enabled
prompts = list(await super().list_prompts())
prompts = await apply_session_transforms(prompts)
prompts = [p for p in prompts if is_enabled(p)]
# Core logic: list prompts
with server_span(
"prompts/list", "prompts/list", self.name, "prompt", ""
):
# Get all prompts, apply session transforms, then filter enabled
prompts = list(await super().list_prompts())
prompts = await apply_session_transforms(prompts)
prompts = [p for p in prompts if is_enabled(p)]
skip_auth, token = _get_auth_context()
authorized: list[Prompt] = []
for prompt in prompts:
if not skip_auth and prompt.auth is not None:
ctx = AuthContext(token=token, component=prompt)
try:
if not await run_auth_checks(prompt.auth, ctx):
skip_auth, token = _get_auth_context()
authorized: list[Prompt] = []
for prompt in prompts:
if not skip_auth and prompt.auth is not None:
ctx = AuthContext(token=token, component=prompt)
try:
if not await run_auth_checks(prompt.auth, ctx):
continue
except AuthorizationError:
continue
except AuthorizationError:
continue
authorized.append(prompt)
return authorized
authorized.append(prompt)
return authorized
async def _get_prompt(
self, name: str, version: VersionSpec | None = None

15
src/fastmcp/server/telemetry.py
View file

@ -93,7 +93,12 @@ def server_span(
try:
yield span
except Exception as e:
span.set_attribute("error.type", type(e).__name__)
span.set_attribute(
"error.type",
"tool_error"
if type(e).__name__ == "ToolError"
else type(e).__name__,
)
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
raise
@ -123,8 +128,14 @@ def delegate_span(
try:
yield span
except Exception as e:
span.set_attribute(
"error.type",
"tool_error"
if type(e).__name__ == "ToolError"
else type(e).__name__,
)
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR))
span.set_status(Status(StatusCode.ERROR, str(e)))
raise

18
tests/client/telemetry/test_client_tracing.py
View file

@ -74,7 +74,7 @@ class TestClientToolTracing:
async def test_call_tool_error_caught_by_client_span(
self, trace_exporter: InMemorySpanExporter
):
"""ToolError from _parse_call_tool_result should be caught by the client span."""
"""Tool error should be reflected on the client span via isError check."""
server = FastMCP("test-server")
@server.tool()
@ -88,7 +88,7 @@ class TestClientToolTracing:
spans = trace_exporter.get_finished_spans()
# Find the outer client span (from call_tool wrapping _parse_call_tool_result)
# Find the client span (from call_tool_mcp)
client_spans = [
s
for s in spans
@ -97,17 +97,15 @@ class TestClientToolTracing:
and "fastmcp.server.name" not in s.attributes
]
# At least one client span should have ERROR status (the one catching ToolError)
error_client_spans = [
s for s in client_spans if s.status.status_code == StatusCode.ERROR
]
assert len(error_client_spans) >= 1, (
"At least one client span should capture the ToolError"
# Exactly one client span should exist (no duplicate from call_tool)
assert len(client_spans) == 1, (
"There should be exactly one client span for call_tool"
)
error_span = error_client_spans[0]
error_span = client_spans[0]
assert error_span.status.status_code == StatusCode.ERROR
assert error_span.attributes is not None
assert error_span.attributes["error.type"] == "ToolError"
assert error_span.attributes["error.type"] == "tool_error"
class TestClientResourceTracing:

3
tests/server/telemetry/test_server_tracing.py
View file

@ -64,7 +64,7 @@ class TestToolTracing:
assert span.status.status_code == StatusCode.ERROR
assert "Something went wrong" in span.status.description
assert span.attributes is not None
assert span.attributes["error.type"] == "ToolError"
assert span.attributes["error.type"] == "tool_error"
assert len(span.events) > 0 # Exception recorded
async def test_call_nonexistent_tool_sets_error(
@ -82,6 +82,7 @@ class TestToolTracing:
assert span.name == "tools/call nonexistent"
assert span.status.status_code == StatusCode.ERROR
assert span.attributes is not None
# NotFoundError is not a ToolError, so uses class name as fallback
assert span.attributes["error.type"] == "NotFoundError"