diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index d872b2b92d..f2c163ee71 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -17,6 +17,7 @@ from functools import partial from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast from opentelemetry import propagate +from opentelemetry import trace as otel_trace from ._tools import FunctionTool from ._types import ( @@ -25,6 +26,11 @@ from ._types import ( Message, ) from .exceptions import ToolException, ToolExecutionException +from .observability import ( + OtelAttr, + create_mcp_client_span, + set_mcp_span_error, +) if sys.version_info >= (3, 11): from typing import Self # pragma: no cover @@ -279,10 +285,21 @@ class MCPTool: self._supports_logging: bool | None = None self._ping_available: bool = True self._pending_reload_tasks: set[asyncio.Task[None]] = set() + self._mcp_protocol_version: str | None = None def __str__(self) -> str: return f"MCPTool(name={self.name}, description={self.description})" + def _mcp_base_span_attributes(self) -> dict[str, Any]: + """Return base MCP span attributes shared across all operations. + + Subclasses override to add transport-specific attributes (server address, port, etc.). + """ + attrs: dict[str, Any] = {} + if self._mcp_protocol_version: + attrs[OtelAttr.MCP_PROTOCOL_VERSION] = self._mcp_protocol_version + return attrs + def _parse_prompt_result_from_mcp( self, mcp_type: types.GetPromptResult, @@ -793,8 +810,12 @@ class MCPTool: inner_exception=ex if isinstance(ex, Exception) else None, ) from ex try: - initialize_result = await session.initialize() - self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) + with create_mcp_client_span("initialize", attributes=self._mcp_base_span_attributes()) as init_span: + initialize_result = await session.initialize() + self._mcp_protocol_version = getattr(initialize_result, "protocolVersion", None) + if self._mcp_protocol_version: + init_span.set_attribute(OtelAttr.MCP_PROTOCOL_VERSION, self._mcp_protocol_version) + self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) except (Exception, asyncio.CancelledError) as ex: if await self._close_and_check_cancelled(ex): raise @@ -812,8 +833,12 @@ class MCPTool: self.session = session elif self.session._request_id == 0: # type: ignore[attr-defined] # If the session is not initialized, we need to reinitialize it - initialize_result = await self.session.initialize() - self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) + with create_mcp_client_span("initialize", attributes=self._mcp_base_span_attributes()) as init_span: + initialize_result = await self.session.initialize() + self._mcp_protocol_version = getattr(initialize_result, "protocolVersion", None) + if self._mcp_protocol_version: + init_span.set_attribute(OtelAttr.MCP_PROTOCOL_VERSION, self._mcp_protocol_version) + self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) elif self._server_capabilities is None: self._set_server_capabilities(getattr(self.session, "_server_capabilities", None)) logger.debug("Connected to MCP server: %s", self.session) @@ -1057,7 +1082,8 @@ class MCPTool: "Skipping MCP prompt loading because the server did not advertise prompts support." ) return - prompt_list = await self.session.list_prompts(params=params) # type: ignore[union-attr] + with create_mcp_client_span("prompts/list", attributes=self._mcp_base_span_attributes()): + prompt_list = await self.session.list_prompts(params=params) # type: ignore[union-attr] break except ClosedResourceError as cl_ex: if attempt == 0: @@ -1142,7 +1168,8 @@ class MCPTool: if not self._supports_tools: logger.debug("Skipping MCP tool loading because the server did not advertise tools support.") return - tool_list = await self.session.list_tools(params=params) # type: ignore[union-attr] + with create_mcp_client_span("tools/list", attributes=self._mcp_base_span_attributes()): + tool_list = await self.session.list_tools(params=params) # type: ignore[union-attr] break except ClosedResourceError as cl_ex: if attempt == 0: @@ -1314,9 +1341,6 @@ class MCPTool: ToolExecutionException: If the MCP server is not connected, tools are not loaded, or the tool call fails. """ - from anyio import ClosedResourceError - from mcp.shared.exceptions import McpError - if not self.load_tools_flag: raise ToolExecutionException( "Tools are not loaded for this server, please set load_tools=True in the constructor." @@ -1365,7 +1389,27 @@ class MCPTool: meta = _inject_otel_into_mcp_meta(request_meta) parser = self.parse_tool_results or self._parse_tool_result_from_mcp - # Try the operation, reconnecting once if the connection is closed + + # Build MCP span attributes for tools/call + mcp_attrs = self._mcp_base_span_attributes() + mcp_attrs[OtelAttr.TOOL_NAME] = tool_name + mcp_attrs[OtelAttr.OPERATION] = OtelAttr.TOOL_EXECUTION_OPERATION + + with create_mcp_client_span("tools/call", target=tool_name, attributes=mcp_attrs) as span: + return await self._call_tool_with_retries(tool_name, filtered_kwargs, meta, parser, span) + + async def _call_tool_with_retries( + self, + tool_name: str, + filtered_kwargs: dict[str, Any], + meta: dict[str, Any] | None, + parser: Callable[..., str | list[Content]], + span: otel_trace.Span | None, + ) -> str | list[Content]: + """Execute the MCP tools/call RPC with retry logic.""" + from anyio import ClosedResourceError + from mcp.shared.exceptions import McpError + for attempt in range(2): try: result = await self.session.call_tool(tool_name, arguments=filtered_kwargs, meta=meta) # type: ignore @@ -1376,18 +1420,34 @@ class MCPTool: if isinstance(parsed, list) else str(parsed) ) + # Per OTel MCP semconv: set error.type="tool_error" for isError results + if span is not None: + set_mcp_span_error(span, "tool_error", text or str(parsed)) raise ToolExecutionException(text or str(parsed)) return parser(result) except ToolExecutionException: raise - except (ClosedResourceError, McpError) as call_ex: + except (ClosedResourceError, McpError) as call_ex: # type: ignore[reportUnknownVariableType] is_session_terminated = ( - isinstance(call_ex, McpError) and "session terminated" in call_ex.error.message.lower() + isinstance(call_ex, McpError) and "session terminated" in call_ex.error.message.lower() # type: ignore[reportUnknownMemberType] ) is_connection_lost = isinstance(call_ex, ClosedResourceError) or is_session_terminated if not is_connection_lost: - error_message = call_ex.error.message if isinstance(call_ex, McpError) else str(call_ex) - raise ToolExecutionException(error_message, inner_exception=call_ex) from call_ex + error_message = ( + call_ex.error.message # type: ignore[reportUnknownMemberType] + if isinstance(call_ex, McpError) + else str(call_ex) # type: ignore[reportUnknownArgumentType] + ) + if span is not None: + set_mcp_span_error( + span, + type(call_ex).__name__, # type: ignore[reportUnknownArgumentType] + error_message, # type: ignore[reportUnknownArgumentType] + ) + raise ToolExecutionException( + error_message, + inner_exception=call_ex, # type: ignore[reportUnknownArgumentType] + ) from call_ex if attempt == 0: # First attempt failed, try reconnecting. @@ -1402,12 +1462,19 @@ class MCPTool: ) from reconn_ex # Second attempt also failed, give up. - logger.error("MCP connection closed unexpectedly after reconnection: %s", call_ex) + logger.error( + "MCP connection closed unexpectedly after reconnection: %s", + call_ex, # type: ignore[reportUnknownArgumentType] + ) + if span is not None: + set_mcp_span_error(span, type(call_ex).__name__, str(call_ex)) # type: ignore[reportUnknownArgumentType] raise ToolExecutionException( f"Failed to call tool '{tool_name}' - connection lost.", - inner_exception=call_ex, + inner_exception=call_ex, # type: ignore[reportUnknownArgumentType] ) from call_ex except Exception as ex: + if span is not None: + set_mcp_span_error(span, type(ex).__name__, str(ex)) raise ToolExecutionException(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex raise ToolExecutionException(f"Failed to call tool '{tool_name}' after retries.") @@ -1436,36 +1503,42 @@ class MCPTool: ) parser = self.parse_prompt_results or self._parse_prompt_result_from_mcp - # Try the operation, reconnecting once if the connection is closed - for attempt in range(2): - try: - prompt_result = await self.session.get_prompt(prompt_name, arguments=kwargs) # type: ignore - return parser(prompt_result) - except ClosedResourceError as cl_ex: - if attempt == 0: - # First attempt failed, try reconnecting - logger.info("MCP connection closed unexpectedly. Reconnecting...") - try: - await self.connect(reset=True) - continue # Retry the operation - except Exception as reconn_ex: + mcp_attrs = self._mcp_base_span_attributes() + mcp_attrs[OtelAttr.PROMPT_NAME] = prompt_name + + with create_mcp_client_span("prompts/get", target=prompt_name, attributes=mcp_attrs) as span: + for attempt in range(2): + try: + prompt_result = await self.session.get_prompt(prompt_name, arguments=kwargs) # type: ignore + return parser(prompt_result) + except ClosedResourceError as cl_ex: + if attempt == 0: + # First attempt failed, try reconnecting + logger.info("MCP connection closed unexpectedly. Reconnecting...") + try: + await self.connect(reset=True) + continue # Retry the operation + except Exception as reconn_ex: + raise ToolExecutionException( + "Failed to reconnect to MCP server.", + inner_exception=reconn_ex, + ) from reconn_ex + else: + # Second attempt also failed, give up + logger.error(f"MCP connection closed unexpectedly after reconnection: {cl_ex}") + set_mcp_span_error(span, type(cl_ex).__name__, str(cl_ex)) raise ToolExecutionException( - "Failed to reconnect to MCP server.", - inner_exception=reconn_ex, - ) from reconn_ex - else: - # Second attempt also failed, give up - logger.error(f"MCP connection closed unexpectedly after reconnection: {cl_ex}") - raise ToolExecutionException( - f"Failed to call prompt '{prompt_name}' - connection lost.", - inner_exception=cl_ex, - ) from cl_ex - except McpError as mcp_exc: - error_message = mcp_exc.error.message - raise ToolExecutionException(error_message, inner_exception=mcp_exc) from mcp_exc - except Exception as ex: - raise ToolExecutionException(f"Failed to call prompt '{prompt_name}'.", inner_exception=ex) from ex - raise ToolExecutionException(f"Failed to get prompt '{prompt_name}' after retries.") + f"Failed to call prompt '{prompt_name}' - connection lost.", + inner_exception=cl_ex, + ) from cl_ex + except McpError as mcp_exc: + error_message = mcp_exc.error.message + set_mcp_span_error(span, type(mcp_exc).__name__, error_message) + raise ToolExecutionException(error_message, inner_exception=mcp_exc) from mcp_exc + except Exception as ex: + set_mcp_span_error(span, type(ex).__name__, str(ex)) + raise ToolExecutionException(f"Failed to call prompt '{prompt_name}'.", inner_exception=ex) from ex + raise ToolExecutionException(f"Failed to get prompt '{prompt_name}' after retries.") async def __aenter__(self) -> Self: """Enter the async context manager. @@ -1621,6 +1694,11 @@ class MCPStdioTool(MCPTool): self.encoding = encoding self._client_kwargs = kwargs + def _mcp_base_span_attributes(self) -> dict[str, Any]: + attrs = super()._mcp_base_span_attributes() + attrs[OtelAttr.NETWORK_TRANSPORT] = "pipe" + return attrs + def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: """Get an MCP stdio client. @@ -1761,6 +1839,24 @@ class MCPStreamableHTTPTool(MCPTool): self._httpx_client: AsyncClient | None = http_client self._header_provider = header_provider + def _mcp_base_span_attributes(self) -> dict[str, Any]: + attrs = super()._mcp_base_span_attributes() + attrs[OtelAttr.NETWORK_TRANSPORT] = "tcp" + attrs[OtelAttr.NETWORK_PROTOCOL_NAME] = "http" + try: + from httpx import URL + + parsed = URL(self.url) + if parsed.host: + attrs[OtelAttr.ADDRESS] = parsed.host + port = parsed.port + if port is None: + port = 443 if parsed.scheme == "https" else 80 + attrs[OtelAttr.PORT] = port + except Exception: + logger.debug("Failed to parse URL for MCP span transport attributes", exc_info=True) + return attrs + def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: """Get an MCP streamable HTTP client. @@ -1924,6 +2020,24 @@ class MCPWebsocketTool(MCPTool): self.url = url self._client_kwargs = kwargs + def _mcp_base_span_attributes(self) -> dict[str, Any]: + attrs = super()._mcp_base_span_attributes() + attrs[OtelAttr.NETWORK_TRANSPORT] = "tcp" + attrs[OtelAttr.NETWORK_PROTOCOL_NAME] = "websocket" + try: + from urllib.parse import urlparse + + parsed = urlparse(self.url) + if parsed.hostname: + attrs[OtelAttr.ADDRESS] = parsed.hostname + port = parsed.port + if port is None: + port = 443 if parsed.scheme == "wss" else 80 + attrs[OtelAttr.PORT] = port + except Exception: + logger.debug("Failed to parse URL for MCP span transport attributes", exc_info=True) + return attrs + def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: """Get an MCP WebSocket client. diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 5237cf62ba..b9e9ecd37f 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -735,7 +735,8 @@ class FunctionTool(SerializationMixin): end_time_stamp = perf_counter() except Exception as exception: end_time_stamp = perf_counter() - attributes[OtelAttr.ERROR_TYPE] = type(exception).__name__ + error_type = type(exception).__name__ + attributes[OtelAttr.ERROR_TYPE] = error_type capture_exception(span=span, exception=exception, timestamp=time_ns()) logger.error(f"Function failed. Error: {exception}") raise diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 1654799d87..ea32ec40ef 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -80,6 +80,7 @@ __all__ = [ "EmbeddingTelemetryLayer", "OtelAttr", "configure_otel_providers", + "create_mcp_client_span", "create_metric_views", "create_resource", "disable_instrumentation", @@ -87,6 +88,8 @@ __all__ = [ "enable_sensitive_telemetry", "get_meter", "get_tracer", + "mcp_tracer", + "set_mcp_span_error", ] @@ -110,7 +113,6 @@ INNER_ACCUMULATED_USAGE: Final[contextvars.ContextVar[UsageDetails | None]] = co "inner_accumulated_usage", default=None ) - OTEL_METRICS: Final[str] = "__otel_metrics__" TOKEN_USAGE_BUCKET_BOUNDARIES: Final[tuple[float, ...]] = ( 1, @@ -292,6 +294,14 @@ class OtelAttr(str, Enum): AGENT_CREATE_OPERATION = "create_agent" AGENT_INVOKE_OPERATION = "invoke_agent" + # MCP attributes (https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp/) + MCP_METHOD_NAME = "mcp.method.name" + MCP_PROTOCOL_VERSION = "mcp.protocol.version" + MCP_SESSION_ID = "mcp.session.id" + PROMPT_NAME = "gen_ai.prompt.name" + NETWORK_TRANSPORT = "network.transport" + NETWORK_PROTOCOL_NAME = "network.protocol.name" + # Agent Framework specific attributes MEASUREMENT_FUNCTION_TAG_NAME = "agent_framework.function.name" MEASUREMENT_FUNCTION_INVOCATION_DURATION = "agent_framework.function.invocation.duration" @@ -2013,6 +2023,65 @@ def get_function_span( ) +# region MCP span helpers + + +def mcp_tracer() -> trace.Tracer: + """Get a tracer for MCP operations, or a no-op tracer if observability is disabled.""" + return get_tracer() if OBSERVABILITY_SETTINGS.ENABLED else trace.NoOpTracer() + + +@contextlib.contextmanager +def create_mcp_client_span( + method_name: str, + target: str | None = None, + attributes: dict[str, Any] | None = None, +) -> Generator[trace.Span, Any, Any]: + """Create an MCP client span per OTel MCP semantic conventions. + + Span name follows the format ``{mcp.method.name} {target}`` when a target + is available, otherwise just ``{mcp.method.name}``. + + See: https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp/#client + + Args: + method_name: The MCP method name (e.g. ``initialize``, ``tools/call``). + target: Optional low-cardinality target (tool name, prompt name). + attributes: Additional span attributes. + """ + span_name = f"{method_name} {target}" if target else method_name + attrs: dict[str, Any] = {OtelAttr.MCP_METHOD_NAME: method_name} + if attributes: + attrs.update(attributes) + span = mcp_tracer().start_span(span_name, kind=trace.SpanKind.CLIENT, attributes=attrs) + with trace.use_span( + span=span, + end_on_exit=True, + record_exception=False, + set_status_on_exception=False, + ) as current_span: + yield current_span + + +def set_mcp_span_error( + span: trace.Span, + error_type: str, + description: str | None = None, +) -> None: + """Set error status and ``error.type`` on an MCP span. + + Args: + span: The span to mark as errored. + error_type: The error type string (e.g. ``tool_error``, exception class name). + description: Optional description (e.g. JSON-RPC error message). + """ + span.set_attribute(OtelAttr.ERROR_TYPE, error_type) + span.set_status(trace.StatusCode.ERROR, description=description) + + +# endregion + + @contextlib.contextmanager def _activate_span(span: trace.Span) -> Generator[None]: """Attach ``span`` as the current span in the OpenTelemetry context. diff --git a/python/packages/core/tests/core/test_mcp_observability.py b/python/packages/core/tests/core/test_mcp_observability.py new file mode 100644 index 0000000000..19d083b317 --- /dev/null +++ b/python/packages/core/tests/core/test_mcp_observability.py @@ -0,0 +1,353 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for MCP client span instrumentation per OTel GenAI Semantic Conventions. + +See: https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp/#client +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest +from mcp import types +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import SpanKind, StatusCode + +from agent_framework import MCPStdioTool, MCPStreamableHTTPTool, MCPWebsocketTool +from agent_framework._mcp import MCPTool +from agent_framework.exceptions import ToolExecutionException +from agent_framework.observability import OtelAttr + +# region helpers + + +def _make_connected_mcp_tool( + name: str = "test-mcp", + *, + supports_tools: bool = True, + supports_prompts: bool = True, + protocol_version: str | None = "2025-06-18", +) -> MCPTool: + """Create an MCPTool with a mocked session, ready for testing.""" + tool = MCPTool(name=name) + tool.session = AsyncMock() + tool.is_connected = True + tool._supports_tools = supports_tools + tool._supports_prompts = supports_prompts + tool.load_tools_flag = True + tool.load_prompts_flag = True + tool._mcp_protocol_version = protocol_version + return tool + + +def _make_tool_list_result( + tools: list[dict[str, Any]] | None = None, +) -> Mock: + """Create a mock ListToolsResult.""" + if tools is None: + tools = [{"name": "get-weather", "description": "Get weather", "inputSchema": {"type": "object"}}] + result = Mock() + result.tools = [ + types.Tool(name=t["name"], description=t.get("description", ""), inputSchema=t.get("inputSchema", {})) + for t in tools + ] + result.nextCursor = None + return result + + +def _make_prompt_list_result( + prompts: list[dict[str, Any]] | None = None, +) -> Mock: + """Create a mock ListPromptsResult.""" + if prompts is None: + prompts = [{"name": "analyze-code", "description": "Analyze code"}] + result = Mock() + result.prompts = [ + types.Prompt(name=p["name"], description=p.get("description", ""), arguments=None) for p in prompts + ] + result.nextCursor = None + return result + + +def _make_call_tool_result(text: str = "result", is_error: bool = False) -> Mock: + """Create a mock CallToolResult.""" + result = Mock() + result.isError = is_error + result.content = [types.TextContent(type="text", text=text)] + return result + + +def _make_get_prompt_result(text: str = "prompt result") -> types.GetPromptResult: + """Create a mock GetPromptResult.""" + return types.GetPromptResult( + description="test prompt", + messages=[ + types.PromptMessage( + role="user", + content=types.TextContent(type="text", text=text), + ) + ], + ) + + +# endregion + + +# region initialize span + + +async def test_mcp_initialize_span(span_exporter: InMemorySpanExporter): + """session.initialize() should produce an MCP CLIENT span named 'initialize'.""" + tool = MCPTool(name="test-server") + + mock_session_cls = AsyncMock() + init_result = Mock() + init_result.capabilities = None + init_result.protocolVersion = "2025-06-18" + mock_session_cls.initialize = AsyncMock(return_value=init_result) + + # Create a mock transport context manager + mock_transport = AsyncMock() + mock_transport.__aenter__ = AsyncMock(return_value=(Mock(), Mock())) + mock_transport.__aexit__ = AsyncMock(return_value=False) + + # Mock get_mcp_client and the session creation + tool.session = None + tool.load_tools_flag = False + tool.load_prompts_flag = False + + span_exporter.clear() + + with pytest.MonkeyPatch.context() as m: + m.setattr(tool, "get_mcp_client", lambda: mock_transport) + + async def patched_connect(self_: Any, *, reset: bool = False, load_configured: bool = True) -> None: + # Simulate what _connect_on_owner does but with mock session + from agent_framework._mcp import create_mcp_client_span + from agent_framework.observability import OtelAttr + + with create_mcp_client_span("initialize", attributes=self_._mcp_base_span_attributes()) as init_span: + self_._mcp_protocol_version = "2025-06-18" + init_span.set_attribute(OtelAttr.MCP_PROTOCOL_VERSION, "2025-06-18") + + self_.session = mock_session_cls + self_.is_connected = True + + m.setattr(MCPTool, "_connect_on_owner", patched_connect) + await tool.connect() + + spans = span_exporter.get_finished_spans() + init_spans = [s for s in spans if s.name == "initialize"] + assert len(init_spans) == 1 + span = init_spans[0] + assert span.kind == SpanKind.CLIENT + assert span.attributes[OtelAttr.MCP_METHOD_NAME] == "initialize" + assert span.attributes.get(OtelAttr.MCP_PROTOCOL_VERSION) == "2025-06-18" + + +# endregion + + +# region tools/list span + + +async def test_mcp_tools_list_span(span_exporter: InMemorySpanExporter): + """session.list_tools() should produce an MCP CLIENT span named 'tools/list'.""" + tool = _make_connected_mcp_tool() + tool.session.list_tools = AsyncMock(return_value=_make_tool_list_result()) + + span_exporter.clear() + await tool.load_tools() + + spans = span_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.CLIENT + assert span.attributes[OtelAttr.MCP_METHOD_NAME] == "tools/list" + assert span.attributes.get(OtelAttr.MCP_PROTOCOL_VERSION) == "2025-06-18" + + +# endregion + + +# region prompts/list span + + +async def test_mcp_prompts_list_span(span_exporter: InMemorySpanExporter): + """session.list_prompts() should produce an MCP CLIENT span named 'prompts/list'.""" + tool = _make_connected_mcp_tool() + tool.session.list_prompts = AsyncMock(return_value=_make_prompt_list_result()) + + span_exporter.clear() + await tool.load_prompts() + + spans = span_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.CLIENT + assert span.attributes[OtelAttr.MCP_METHOD_NAME] == "prompts/list" + + +# endregion + + +# region tools/call span + + +async def test_mcp_tools_call_creates_client_span_when_no_parent(span_exporter: InMemorySpanExporter): + """Direct call_tool() without FunctionTool wrapper creates new MCP CLIENT span.""" + tool = _make_connected_mcp_tool() + tool.session.call_tool = AsyncMock(return_value=_make_call_tool_result("hello")) + + span_exporter.clear() + result = await tool.call_tool("get-weather", city="Seattle") + + assert result is not None + spans = span_exporter.get_finished_spans() + call_spans = [s for s in spans if "tools/call" in s.name] + assert len(call_spans) == 1 + span = call_spans[0] + assert span.kind == SpanKind.CLIENT + assert span.name == "tools/call get-weather" + assert span.attributes[OtelAttr.MCP_METHOD_NAME] == "tools/call" + assert span.attributes[OtelAttr.TOOL_NAME] == "get-weather" + assert span.attributes.get(OtelAttr.MCP_PROTOCOL_VERSION) == "2025-06-18" + + +async def test_mcp_tools_call_tool_error_sets_error_type(span_exporter: InMemorySpanExporter): + """When CallToolResult.isError is true, error.type should be 'tool_error' per MCP spec.""" + tool = _make_connected_mcp_tool() + tool.session.call_tool = AsyncMock(return_value=_make_call_tool_result("bad input", is_error=True)) + + span_exporter.clear() + with pytest.raises(ToolExecutionException): + await tool.call_tool("get-weather", city="invalid") + + spans = span_exporter.get_finished_spans() + call_spans = [s for s in spans if "tools/call" in s.name] + assert len(call_spans) == 1 + span = call_spans[0] + assert span.attributes.get(OtelAttr.ERROR_TYPE) == "tool_error" + assert span.status.status_code == StatusCode.ERROR + + +# endregion + + +# region prompts/get span + + +async def test_mcp_prompts_get_creates_client_span(span_exporter: InMemorySpanExporter): + """get_prompt() should always create a new MCP CLIENT span (not enrich execute_tool).""" + tool = _make_connected_mcp_tool() + tool.session.get_prompt = AsyncMock(return_value=_make_get_prompt_result("code analysis")) + + span_exporter.clear() + result = await tool.get_prompt("analyze-code", language="python") + + assert "code analysis" in result + spans = span_exporter.get_finished_spans() + prompt_spans = [s for s in spans if "prompts/get" in s.name] + assert len(prompt_spans) == 1 + span = prompt_spans[0] + assert span.kind == SpanKind.CLIENT + assert span.name == "prompts/get analyze-code" + assert span.attributes[OtelAttr.MCP_METHOD_NAME] == "prompts/get" + assert span.attributes[OtelAttr.PROMPT_NAME] == "analyze-code" + + +# endregion + + +# region transport attributes + + +def test_mcp_stdio_tool_transport_attributes(): + """MCPStdioTool should have network.transport='pipe'.""" + tool = MCPStdioTool(name="test", command="python") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.NETWORK_TRANSPORT] == "pipe" + assert OtelAttr.ADDRESS not in attrs + + +def test_mcp_http_tool_transport_attributes(): + """MCPStreamableHTTPTool should have tcp transport and URL-based server address/port.""" + tool = MCPStreamableHTTPTool(name="test", url="https://api.example.com:8443/mcp") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.NETWORK_TRANSPORT] == "tcp" + assert attrs[OtelAttr.NETWORK_PROTOCOL_NAME] == "http" + assert attrs[OtelAttr.ADDRESS] == "api.example.com" + assert attrs[OtelAttr.PORT] == 8443 + + +def test_mcp_http_tool_default_port(): + """MCPStreamableHTTPTool should default to 443 for https.""" + tool = MCPStreamableHTTPTool(name="test", url="https://api.example.com/mcp") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.PORT] == 443 + + +def test_mcp_http_tool_http_default_port(): + """MCPStreamableHTTPTool should default to 80 for http.""" + tool = MCPStreamableHTTPTool(name="test", url="http://localhost/mcp") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.PORT] == 80 + + +def test_mcp_websocket_tool_transport_attributes(): + """MCPWebsocketTool should have tcp transport and URL-based server address/port.""" + tool = MCPWebsocketTool(name="test", url="wss://ws.example.com:9090/mcp") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.NETWORK_TRANSPORT] == "tcp" + assert attrs[OtelAttr.NETWORK_PROTOCOL_NAME] == "websocket" + assert attrs[OtelAttr.ADDRESS] == "ws.example.com" + assert attrs[OtelAttr.PORT] == 9090 + + +def test_mcp_websocket_tool_default_port(): + """MCPWebsocketTool should default to 443 for wss.""" + tool = MCPWebsocketTool(name="test", url="wss://ws.example.com/mcp") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.PORT] == 443 + + +def test_mcp_protocol_version_in_attributes(): + """Protocol version should be included in span attributes.""" + tool = _make_connected_mcp_tool(protocol_version="2025-06-18") + attrs = tool._mcp_base_span_attributes() + assert attrs[OtelAttr.MCP_PROTOCOL_VERSION] == "2025-06-18" + + +def test_mcp_protocol_version_omitted_when_none(): + """Protocol version should be omitted from span attributes when unknown.""" + tool = _make_connected_mcp_tool(protocol_version=None) + attrs = tool._mcp_base_span_attributes() + assert OtelAttr.MCP_PROTOCOL_VERSION not in attrs + + +# endregion + + +# region observability disabled + + +@pytest.mark.parametrize("enable_instrumentation", [False], indirect=True) +async def test_mcp_spans_not_created_when_observability_disabled(span_exporter: InMemorySpanExporter): + """No MCP spans should be created when observability is disabled.""" + tool = _make_connected_mcp_tool() + tool.session.list_tools = AsyncMock(return_value=_make_tool_list_result()) + tool.session.call_tool = AsyncMock(return_value=_make_call_tool_result("ok")) + + span_exporter.clear() + await tool.load_tools() + await tool.call_tool("get-weather", city="Seattle") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 0 + + +# endregion