diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 89d32f9000d..259f3c2dba8 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -75,6 +75,68 @@ logger = logging.getLogger(__name__) + +def _mcp_exception_log_level() -> Literal[10] | None: + """Return the automatic exception log level allowed by observability settings.""" + from .observability import OBSERVABILITY_SETTINGS + + return 10 if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED else None + + +def _log_mcp_exception( + level: int, + exception: BaseException, + sensitive_message: str, + *sensitive_args: object, + redacted_message: str, + sensitive_exc_info: bool | BaseException = False, +) -> None: + """Log an MCP exception without exporting details unless explicitly enabled.""" + from .observability import OBSERVABILITY_SETTINGS + + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + message = sensitive_message + args = sensitive_args + exc_info = sensitive_exc_info + else: + message = "%s (%s)." + args = (redacted_message, type(exception).__name__) + exc_info = False + + if level == logging.DEBUG: + logger.debug(message, *args, exc_info=exc_info) + elif level == logging.WARNING: + logger.warning(message, *args, exc_info=exc_info) + elif level == logging.ERROR: + logger.error(message, *args, exc_info=exc_info) + else: + logger.log(level, message, *args, exc_info=exc_info) + + +def _mcp_tool_exception( + message: str, + inner_exception: Exception | None = None, +) -> ToolException: + """Create an MCP tool lifecycle exception without leaking details through automatic logs.""" + return ToolException( + message, + inner_exception=inner_exception, + log_level=_mcp_exception_log_level(), + ) + + +def _mcp_tool_execution_exception( + message: str, + inner_exception: Exception | None = None, +) -> ToolExecutionException: + """Create an MCP tool exception without leaking details through automatic logs.""" + return ToolExecutionException( + message, + inner_exception=inner_exception, + log_level=_mcp_exception_log_level(), + ) + + if TYPE_CHECKING: _MCPSamplingContentBlock: TypeAlias = ( types.TextContent @@ -669,7 +731,7 @@ async def _call_prompt_with_runtime_kwargs( def _validate_mcp_meta_key(key: str) -> None: """Validate an MCP ``_meta`` key against the 2025-06-18 key-name format.""" if not _MCP_META_KEY_PATTERN.fullmatch(key): - raise ToolExecutionException(f"Invalid MCP _meta key name: {key!r}.") + raise _mcp_tool_execution_exception(f"Invalid MCP _meta key name: {key!r}.") def _validate_mcp_meta(raw_meta: object | None) -> dict[str, Any] | None: @@ -677,13 +739,13 @@ def _validate_mcp_meta(raw_meta: object | None) -> dict[str, Any] | None: if raw_meta is None: return None if not isinstance(raw_meta, dict): - raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.") + raise _mcp_tool_execution_exception("MCP tool metadata provided via _meta must be a dict.") raw_meta_dict = cast(Mapping[object, Any], raw_meta) meta: dict[str, Any] = {} for key, value in raw_meta_dict.items(): if not isinstance(key, str): - raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.") + raise _mcp_tool_execution_exception("MCP tool metadata provided via _meta must use string keys.") _validate_mcp_meta_key(key) meta[key] = value return meta @@ -746,6 +808,14 @@ class _MCPTaskAbandoned(ToolExecutionException): Subclass of ToolExecutionException so callers see a normal tool failure. """ + def __init__(self, message: str, inner_exception: Exception | None = None) -> None: + """Create an abandonment exception without leaking details through automatic logs.""" + super().__init__( + message, + inner_exception=inner_exception, + log_level=_mcp_exception_log_level(), + ) + class _MCPDeadlineExpired(Exception): """Internal marker for ``max_task_wait`` expiry; distinct from inner TimeoutError.""" @@ -1442,7 +1512,7 @@ def _validate_config_names(self, functions: Sequence[FunctionTool]) -> None: continue existing_remote = remote_by_config_name.get(name) if existing_remote is not None and existing_remote != remote_name: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"MCP configuration name {name!r} is ambiguous: it matches raw remote names " f"{existing_remote!r} and {remote_name!r}. " "Use an unambiguous name or a different tool_name_prefix." @@ -1561,9 +1631,11 @@ def _resolve_progressive_function(self, tool_name: str) -> FunctionTool: ) or "none" ) - raise ToolExecutionException(f"MCP tool '{tool_name}' is not available. Available tools: {available}.") + raise _mcp_tool_execution_exception( + f"MCP tool '{tool_name}' is not available. Available tools: {available}." + ) if len(matches) > 1: - raise ToolExecutionException(f"MCP tool name '{tool_name}' is ambiguous.") + raise _mcp_tool_execution_exception(f"MCP tool name '{tool_name}' is ambiguous.") return matches[0] @staticmethod @@ -1572,10 +1644,10 @@ def _progressive_tool_names(tool: str | Sequence[str]) -> list[str]: if isinstance(tool, str): return [tool] if not isinstance(tool, Sequence): - raise ToolExecutionException("Progressive MCP tool request must be a string or a list of strings.") + raise _mcp_tool_execution_exception("Progressive MCP tool request must be a string or a list of strings.") tool_names = list(tool) if not all(isinstance(tool_name, str) for tool_name in tool_names): - raise ToolExecutionException("Progressive MCP tool request must contain only strings.") + raise _mcp_tool_execution_exception("Progressive MCP tool request must contain only strings.") return tool_names def _is_progressive_function_loaded( @@ -1614,7 +1686,7 @@ def _list_progressive_mcp_tools(self, ctx: FunctionInvocationContext) -> list[di async def _load_progressive_mcp_tool(self, ctx: FunctionInvocationContext, tool: str | Sequence[str]) -> str: """Load an allowed MCP tool into the live function-calling tool list.""" if ctx.tools is None: - raise ToolExecutionException("load_tool can only be used inside an agent function-calling run.") + raise _mcp_tool_execution_exception("load_tool can only be used inside an agent function-calling run.") messages: list[str] = [] functions_to_load: list[FunctionTool] = [] function_names_to_load: set[str] = set() @@ -1650,14 +1722,14 @@ async def _load_progressive_mcp_tool(self, ctx: FunctionInvocationContext, tool: warnings.simplefilter("ignore", ExperimentalWarning) ctx.add_tools(functions_to_load) except ValueError as ex: - raise ToolExecutionException(str(ex), inner_exception=ex) from ex + raise _mcp_tool_execution_exception(str(ex), inner_exception=ex) from ex self._progressive_loaded_tool_names.update(func.name for func in functions_to_load) return "\n".join(messages) async def _unload_progressive_mcp_tool(self, ctx: FunctionInvocationContext, tool: str | Sequence[str]) -> str: """Unload a progressively loaded MCP tool from the live function-calling tool list.""" if ctx.tools is None: - raise ToolExecutionException("unload_tool can only be used inside an agent function-calling run.") + raise _mcp_tool_execution_exception("unload_tool can only be used inside an agent function-calling run.") messages: list[str] = [] function_names_to_unload: set[str] = set() for tool_name in self._progressive_tool_names(tool): @@ -1698,7 +1770,7 @@ async def _unload_progressive_mcp_tool(self, ctx: FunctionInvocationContext, too warnings.simplefilter("ignore", ExperimentalWarning) ctx.remove_tools(list(function_names_to_unload)) except RuntimeError as ex: - raise ToolExecutionException(str(ex), inner_exception=ex) from ex + raise _mcp_tool_execution_exception(str(ex), inner_exception=ex) from ex self._progressive_loaded_tool_names.difference_update(function_names_to_unload) return "\n".join(messages) @@ -1764,8 +1836,13 @@ async def _run_lifecycle_owner(self) -> None: if not future.done(): future.set_exception(ex) else: - logger.warning( - "MCP lifecycle action %s failed after its caller stopped waiting.", action, exc_info=ex + _log_mcp_exception( + logging.WARNING, + ex, + "MCP lifecycle action %s failed after its caller stopped waiting.", + action, + redacted_message="MCP lifecycle action failed after its caller stopped waiting", + sensitive_exc_info=ex, ) # A connect that failed without leaving a session behind has nothing for this # owner to hold, so stop instead of blocking on the queue forever. Mirrors the @@ -1849,10 +1926,13 @@ async def _safe_close_exit_stack(self) -> BaseException | None: except RuntimeError as e: error_msg = str(e).lower() if "cancel scope" in error_msg: - logger.warning( + _log_mcp_exception( + logging.WARNING, + e, "Could not cleanly close MCP exit stack due to cancel scope error. " "This indicates MCP lifecycle ownership was lost. Error: %s", e, + redacted_message="Could not cleanly close MCP exit stack due to cancel scope error", ) return e raise @@ -1861,7 +1941,13 @@ async def _safe_close_exit_stack(self) -> BaseException | None: return e except Exception as e: if type(e).__name__ == "ExceptionGroup": - logger.warning("Could not cleanly close MCP exit stack due to cleanup error group. Error: %s", e) + _log_mcp_exception( + logging.WARNING, + e, + "Could not cleanly close MCP exit stack due to cleanup error group. Error: %s", + e, + redacted_message="Could not cleanly close MCP exit stack due to cleanup error group", + ) return e raise return None @@ -1987,15 +2073,24 @@ async def _connect_on_owner( # CancelledError is a BaseException (not Exception) on Python >= 3.8, so # inner_exception=None and ToolException.__init__ won't log exc_info. if isinstance(ex, asyncio.CancelledError): - logger.debug(error_msg, exc_info=True) - raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex + _log_mcp_exception( + logging.DEBUG, + ex, + error_msg, + redacted_message="MCP transport startup failed", + sensitive_exc_info=True, + ) + raise _mcp_tool_exception( + error_msg, + inner_exception=ex if isinstance(ex, Exception) else None, + ) from ex try: try: from mcp import types from mcp.client.session import ClientSession as runtime_client_session except ModuleNotFoundError as ex: await self._safe_close_exit_stack() - raise ToolException( + raise _mcp_tool_exception( "MCP support requires `mcp`. Please install `mcp`.", inner_exception=ex, ) from ex @@ -2024,14 +2119,28 @@ async def _connect_on_owner( raise session_error_msg = f"Failed to create MCP session: {_describe_with_cleanup(ex, cleanup_error)}" if isinstance(ex, asyncio.CancelledError): - logger.debug(session_error_msg, exc_info=True) - raise ToolException( + _log_mcp_exception( + logging.DEBUG, + ex, + session_error_msg, + redacted_message="MCP session creation failed", + sensitive_exc_info=True, + ) + raise _mcp_tool_exception( message=session_error_msg, inner_exception=ex if isinstance(ex, Exception) else None, ) from ex try: - with create_mcp_client_span("initialize", attributes=self._mcp_base_span_attributes()) as init_span: - initialize_result = await session.initialize() + with create_mcp_client_span( + "initialize", + attributes=self._mcp_base_span_attributes(), + capture_unhandled_exceptions=False, + ) as init_span: + try: + initialize_result = await session.initialize() + except Exception as init_ex: + set_mcp_span_error(init_span, type(init_ex).__name__, str(init_ex)) + raise init_span.set_attribute(OtelAttr.MCP_PROTOCOL_VERSION, initialize_result.protocolVersion) self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) except (Exception, asyncio.CancelledError) as ex: @@ -2048,16 +2157,33 @@ async def _connect_on_owner( else: error_msg = f"MCP server failed to initialize: {_describe_with_cleanup(ex, cleanup_error)}" if isinstance(ex, asyncio.CancelledError): - logger.debug(error_msg, exc_info=True) - raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex + _log_mcp_exception( + logging.DEBUG, + ex, + error_msg, + redacted_message="MCP session initialization failed", + sensitive_exc_info=True, + ) + raise _mcp_tool_exception( + error_msg, + inner_exception=ex if isinstance(ex, Exception) else None, + ) from ex self.session = session self._owns_session = True else: try: if self.session._request_id == 0: # type: ignore[attr-defined] # If the session is not initialized, we need to reinitialize it - with create_mcp_client_span("initialize", attributes=self._mcp_base_span_attributes()) as init_span: - initialize_result = await self.session.initialize() + with create_mcp_client_span( + "initialize", + attributes=self._mcp_base_span_attributes(), + capture_unhandled_exceptions=False, + ) as init_span: + try: + initialize_result = await self.session.initialize() + except Exception as init_ex: + set_mcp_span_error(init_span, type(init_ex).__name__, str(init_ex)) + raise init_span.set_attribute(OtelAttr.MCP_PROTOCOL_VERSION, initialize_result.protocolVersion) self._set_server_capabilities(getattr(initialize_result, "capabilities", None)) elif self._server_capabilities is None: @@ -2088,7 +2214,14 @@ async def _connect_on_owner( ) await self.session.set_logging_level(level_name) except Exception as exc: - logger.warning("Failed to set log level to %s", logger.level, exc_info=exc) + _log_mcp_exception( + logging.WARNING, + exc, + "Failed to set log level to %s", + logger.level, + redacted_message="Failed to set MCP server log level", + sensitive_exc_info=exc, + ) except (Exception, asyncio.CancelledError): try: await self._close_on_owner() @@ -2129,11 +2262,14 @@ async def _sampling_request_approved(self, params: types.CreateMessageRequestPar if isawaitable(outcome): outcome = await outcome except Exception as ex: - logger.warning( + _log_mcp_exception( + logging.WARNING, + ex, "Denying MCP sampling request from '%s': approval callback raised %s.", self.name, ex, - exc_info=True, + redacted_message="Denied MCP sampling request because the approval callback failed", + sensitive_exc_info=True, ) return False approved = bool(outcome) @@ -2277,7 +2413,14 @@ async def sampling_callback( options=options or None, ) except Exception as ex: - logger.debug("Sampling callback error: %s", ex, exc_info=True) + _log_mcp_exception( + logging.DEBUG, + ex, + "Sampling callback error: %s", + ex, + redacted_message="MCP sampling callback failed", + sensitive_exc_info=True, + ) return types.ErrorData( code=types.INTERNAL_ERROR, message=f"Failed to get chat message content: {ex}", @@ -2353,7 +2496,14 @@ async def message_handler( from mcp import types if isinstance(message, Exception): - logger.error("Error from MCP server: %s", message, exc_info=message) + _log_mcp_exception( + logging.ERROR, + message, + "Error from MCP server: %s", + message, + redacted_message="MCP server reported an error", + sensitive_exc_info=message, + ) return if isinstance(message, types.ServerNotification): match message.root.method: @@ -2392,8 +2542,14 @@ async def _safe_reload() -> None: await coro except asyncio.CancelledError: raise - except Exception: - logger.warning("Background MCP reload failed", exc_info=True) + except Exception as ex: + _log_mcp_exception( + logging.WARNING, + ex, + "Background MCP reload failed", + redacted_message="Background MCP reload failed", + sensitive_exc_info=True, + ) task = asyncio.create_task(_safe_reload(), name=reload_name) self._pending_reload_tasks.add(task) @@ -2472,8 +2628,16 @@ async def _load_prompts_locked(self) -> None: "Skipping MCP prompt loading because the server did not advertise prompts support." ) return - 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] + with create_mcp_client_span( + "prompts/list", + attributes=self._mcp_base_span_attributes(), + capture_unhandled_exceptions=False, + ) as list_span: + try: + prompt_list = await self.session.list_prompts(params=params) # type: ignore[union-attr] + except Exception as list_ex: + set_mcp_span_error(list_span, type(list_ex).__name__, str(list_ex)) + raise break except ClosedResourceError as cl_ex: if attempt == 0: @@ -2481,19 +2645,25 @@ async def _load_prompts_locked(self) -> None: try: await self._reconnect_without_loading() except Exception as reconn_ex: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Failed to reconnect to MCP server.", inner_exception=reconn_ex, ) from reconn_ex continue - logger.error("MCP connection closed unexpectedly after reconnection: %s", cl_ex) - raise ToolExecutionException( + _log_mcp_exception( + logging.ERROR, + cl_ex, + "MCP connection closed unexpectedly after reconnection: %s", + cl_ex, + redacted_message="MCP prompt discovery connection closed after reconnection", + ) + raise _mcp_tool_execution_exception( "Failed to load prompts - connection lost.", inner_exception=cl_ex, ) from cl_ex if prompt_list is None: - raise ToolExecutionException("Failed to load prompts.") + raise _mcp_tool_execution_exception("Failed to load prompts.") for prompt in prompt_list.prompts: normalized_name = _normalize_mcp_name(prompt.name) @@ -2605,8 +2775,16 @@ async def _load_tools_locked(self) -> None: if not self._supports_tools: logger.debug("Skipping MCP tool loading because the server did not advertise tools support.") return - 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] + with create_mcp_client_span( + "tools/list", + attributes=self._mcp_base_span_attributes(), + capture_unhandled_exceptions=False, + ) as list_span: + try: + tool_list = await self.session.list_tools(params=params) # type: ignore[union-attr] + except Exception as list_ex: + set_mcp_span_error(list_span, type(list_ex).__name__, str(list_ex)) + raise break except ClosedResourceError as cl_ex: if attempt == 0: @@ -2614,19 +2792,25 @@ async def _load_tools_locked(self) -> None: try: await self._reconnect_without_loading() except Exception as reconn_ex: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Failed to reconnect to MCP server.", inner_exception=reconn_ex, ) from reconn_ex continue - logger.error("MCP connection closed unexpectedly after reconnection: %s", cl_ex) - raise ToolExecutionException( + _log_mcp_exception( + logging.ERROR, + cl_ex, + "MCP connection closed unexpectedly after reconnection: %s", + cl_ex, + redacted_message="MCP tool discovery connection closed after reconnection", + ) + raise _mcp_tool_execution_exception( "Failed to load tools - connection lost.", inner_exception=cl_ex, ) from cl_ex if tool_list is None: - raise ToolExecutionException("Failed to load tools.") + raise _mcp_tool_execution_exception("Failed to load tools.") for tool in tool_list.tools: tool_annotations_by_name[tool.name] = tool.annotations @@ -2661,7 +2845,7 @@ async def _load_tools_locked(self) -> None: # Skip if already loaded if local_name in existing_remote_by_local: if existing_remote_by_local.get(local_name) != tool.name: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "MCP server advertised multiple tools that map to the same local function name: " f"{existing_remote_by_local[local_name]!r} and {tool.name!r} both map to " f"{local_name!r}." @@ -2795,7 +2979,7 @@ async def _ensure_connected(self) -> None: try: await self._reconnect_without_loading() except Exception as ex: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Failed to establish MCP connection.", inner_exception=ex, ) from ex @@ -2804,7 +2988,7 @@ async def _ensure_connected(self) -> None: try: await self._reconnect_without_loading() except Exception as ex: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Failed to establish MCP connection.", inner_exception=ex, ) from ex @@ -2867,7 +3051,7 @@ async def _call_tool( kwargs: dict[str, Any], ) -> str | list[Content]: if not self.load_tools_flag: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Tools are not loaded for this server, please set load_tools=True in the constructor." ) @@ -2886,7 +3070,12 @@ async def _call_tool( OtelAttr.TOOL_NAME: tool_name, OtelAttr.OPERATION: OtelAttr.TOOL_EXECUTION_OPERATION, }) - with create_mcp_client_span("tools/call", target=tool_name, attributes=mcp_span_attrs) as span: + with create_mcp_client_span( + "tools/call", + target=tool_name, + attributes=mcp_span_attrs, + capture_unhandled_exceptions=False, + ) as span: return await self._call_tool_with_retries( tool_name, filtered_kwargs, @@ -2908,6 +3097,7 @@ async def _call_tool_with_retries( from mcp.shared.exceptions import McpError for attempt in range(2): + span_error_recorded = False try: result = await self.session.call_tool(tool_name, arguments=filtered_kwargs, meta=meta) # type: ignore _capture_mcp_tool_result(result) @@ -2921,9 +3111,12 @@ async def _call_tool_with_retries( # Per OTel MCP semconv: set error.type="tool_error" for isError results if span.is_recording(): set_mcp_span_error(span, "tool_error", text or str(parsed)) - raise ToolExecutionException(text or str(parsed)) + span_error_recorded = True + raise _mcp_tool_execution_exception(text or str(parsed)) return parser(result) - except ToolExecutionException: + except ToolExecutionException as tool_ex: + if span.is_recording() and not span_error_recorded: + set_mcp_span_error(span, type(tool_ex).__name__, str(tool_ex)) raise except (ClosedResourceError, McpError) as call_ex: is_session_terminated = ( @@ -2934,7 +3127,7 @@ async def _call_tool_with_retries( error_message = call_ex.error.message if isinstance(call_ex, McpError) else str(call_ex) if span.is_recording(): set_mcp_span_error(span, type(call_ex).__name__, error_message) - raise ToolExecutionException(error_message, inner_exception=call_ex) from call_ex + raise _mcp_tool_execution_exception(error_message, inner_exception=call_ex) from call_ex if attempt == 0: # First attempt failed, try reconnecting. @@ -2943,24 +3136,35 @@ async def _call_tool_with_retries( await self.connect(reset=True) continue except Exception as reconn_ex: - raise ToolExecutionException( + if span.is_recording(): + set_mcp_span_error(span, type(reconn_ex).__name__, str(reconn_ex)) + raise _mcp_tool_execution_exception( "Failed to reconnect to MCP server.", inner_exception=reconn_ex, ) from reconn_ex # Second attempt also failed, give up. - logger.error("MCP connection closed unexpectedly after reconnection: %s", call_ex) + _log_mcp_exception( + logging.ERROR, + call_ex, + "MCP connection closed unexpectedly after reconnection: %s", + call_ex, + redacted_message="MCP tool call connection closed after reconnection", + ) if span.is_recording(): set_mcp_span_error(span, type(call_ex).__name__, str(call_ex)) - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"Failed to call tool '{tool_name}' - connection lost.", inner_exception=call_ex, ) from call_ex except Exception as ex: if span.is_recording(): 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.") + raise _mcp_tool_execution_exception(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex + retries_exhausted = _mcp_tool_execution_exception(f"Failed to call tool '{tool_name}' after retries.") + if span.is_recording(): + set_mcp_span_error(span, type(retries_exhausted).__name__, str(retries_exhausted)) + raise retries_exhausted def _resolved_extra_args(self, tool_name: str) -> set[str]: """Return the user-configured extra argument names allowed for a tool.""" @@ -3043,7 +3247,7 @@ async def _call_tool_as_task( from mcp.shared.exceptions import McpError if not self.load_tools_flag: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Tools are not loaded for this server, please set load_tools=True in the constructor." ) @@ -3059,15 +3263,15 @@ async def _call_tool_as_task( except (ClosedResourceError, McpError) as ex: if not self._is_connection_lost(ex): error_message = ex.error.message if isinstance(ex, McpError) else str(ex) - raise ToolExecutionException(error_message, inner_exception=ex) from ex - raise ToolExecutionException( + raise _mcp_tool_execution_exception(error_message, inner_exception=ex) from ex + raise _mcp_tool_execution_exception( f"Failed to call tool '{tool_name}' - connection lost; task state unknown.", inner_exception=ex, ) from ex except ToolExecutionException: raise except Exception as ex: - raise ToolExecutionException(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex + raise _mcp_tool_execution_exception(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex # Server returned a CallToolResult (no task created) or fell back to plain tools/call. if fallback_result is not None: @@ -3079,11 +3283,13 @@ async def _call_tool_as_task( if isinstance(parsed, list) else str(parsed) ) - raise ToolExecutionException(text or str(parsed)) + raise _mcp_tool_execution_exception(text or str(parsed)) return parser(fallback_result) if task_id is None: - raise ToolExecutionException(f"MCP server did not return a task_id or fallback result for '{tool_name}'.") + raise _mcp_tool_execution_exception( + f"MCP server did not return a task_id or fallback result for '{tool_name}'." + ) # Track to completion: poll until terminal, then fetch payload. Never re-issue # tools/call past this point; reconnect-and-retry only against the same task_id. @@ -3106,7 +3312,7 @@ async def _await_task_completion() -> str | list[Content]: return cast("str | list[Content]", result) except _MCPDeadlineExpired as ex: self._spawn_best_effort_cancel(task_id) - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"MCP task '{task_id}' exceeded max_task_wait of {max_wait_s}s.", inner_exception=ex, ) from ex @@ -3189,7 +3395,7 @@ async def _call_tool_as_task_create( except ValidationError as ex: # Augmented call succeeded server-side; re-issuing a plain tools/call # could double-execute a side-effecting tool. - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"MCP server returned an unparseable response to augmented tools/call " f"for '{tool_name}'; cannot safely retry (server may have started the operation).", inner_exception=ex, @@ -3243,7 +3449,7 @@ def _coerce_get_task_result(lenient: types.Result, task_id: str) -> types.GetTas try: return types.GetTaskResult.model_validate(raw) except Exception as ex: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"MCP server returned a malformed tasks/get response for task '{task_id}'.", inner_exception=ex, ) from ex @@ -3279,7 +3485,7 @@ async def _handle_terminal_task( if isinstance(parsed, list) else str(parsed) ) - raise ToolExecutionException(text or str(parsed)) + raise _mcp_tool_execution_exception(text or str(parsed)) return parser(payload) # Non-completed terminal statuses surface as ToolExecutionException so the @@ -3289,7 +3495,7 @@ async def _handle_terminal_task( # Spec-non-terminal; treated as terminal here because the framework does # not implement the interactive input flow. message = snapshot.statusMessage or "MCP task requires additional input and cannot continue." - raise ToolExecutionException(f"Tool '{tool_name}' task {status}: {message}") + raise _mcp_tool_execution_exception(f"Tool '{tool_name}' task {status}: {message}") async def _fetch_task_result(self, task_id: str) -> types.CallToolResult: """Send ``tasks/result`` and reinterpret the open-typed payload as a CallToolResult.""" @@ -3308,7 +3514,7 @@ async def _fetch_task_result(self, task_id: str) -> types.CallToolResult: ) except McpError as ex: # Server reported completed; a hard fetch error is a plain failure (no cancel). - raise ToolExecutionException(ex.error.message, inner_exception=ex) from ex + raise _mcp_tool_execution_exception(ex.error.message, inner_exception=ex) from ex # GetTaskPayloadResult carries the tool result via extra fields; reinterpret as CallToolResult. payload_dict = payload.model_dump(by_alias=True, exclude_none=True) @@ -3316,7 +3522,7 @@ async def _fetch_task_result(self, task_id: str) -> types.CallToolResult: return types.CallToolResult.model_validate(payload_dict) except ValidationError as ex: # Server reported completed; malformed payload is a plain failure (no cancel needed). - raise ToolExecutionException( + raise _mcp_tool_execution_exception( f"MCP task '{task_id}' result payload could not be parsed as a CallToolResult.", inner_exception=ex, ) from ex @@ -3417,11 +3623,14 @@ async def _try_cancel_task(self, task_id: str) -> None: task_id, _MCP_TASK_CANCEL_TIMEOUT.total_seconds(), ) - except Exception: - logger.warning( + except Exception as ex: + _log_mcp_exception( + logging.WARNING, + ex, "Best-effort tasks/cancel for '%s' failed; remote task may still be running.", task_id, - exc_info=True, + redacted_message="Best-effort MCP task cancellation failed", + sensitive_exc_info=True, ) @staticmethod @@ -3465,7 +3674,7 @@ async def get_prompt(self, prompt_name: str, **kwargs: Any) -> str: from mcp.shared.exceptions import McpError if not self.load_prompts_flag: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "Prompts are not loaded for this server, please set load_prompts=True in the constructor." ) @@ -3473,7 +3682,12 @@ async def get_prompt(self, prompt_name: str, **kwargs: Any) -> str: mcp_span_attrs = self._mcp_base_span_attributes() mcp_span_attrs.update({OtelAttr.PROMPT_NAME: prompt_name}) - with create_mcp_client_span("prompts/get", target=prompt_name, attributes=mcp_span_attrs) as span: + with create_mcp_client_span( + "prompts/get", + target=prompt_name, + attributes=mcp_span_attrs, + capture_unhandled_exceptions=False, + ) as span: for attempt in range(2): try: prompt_result = await self.session.get_prompt(prompt_name, arguments=kwargs) # type: ignore @@ -3486,26 +3700,37 @@ async def get_prompt(self, prompt_name: str, **kwargs: Any) -> str: await self.connect(reset=True) continue # Retry the operation except Exception as reconn_ex: - raise ToolExecutionException( + set_mcp_span_error(span, type(reconn_ex).__name__, str(reconn_ex)) + raise _mcp_tool_execution_exception( "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}") + _log_mcp_exception( + logging.ERROR, + cl_ex, + "MCP connection closed unexpectedly after reconnection: %s", + cl_ex, + redacted_message="MCP prompt call connection closed after reconnection", + ) set_mcp_span_error(span, type(cl_ex).__name__, str(cl_ex)) - raise ToolExecutionException( + raise _mcp_tool_execution_exception( 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 + raise _mcp_tool_execution_exception(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.") + raise _mcp_tool_execution_exception( + f"Failed to call prompt '{prompt_name}'.", inner_exception=ex + ) from ex + retries_exhausted = _mcp_tool_execution_exception(f"Failed to get prompt '{prompt_name}' after retries.") + set_mcp_span_error(span, type(retries_exhausted).__name__, str(retries_exhausted)) + raise retries_exhausted async def __aenter__(self) -> Self: """Enter the async context manager. @@ -3526,7 +3751,7 @@ async def __aenter__(self) -> Self: raise except Exception as ex: await self.close() - raise ToolExecutionException("Failed to enter context manager.", inner_exception=ex) from ex + raise _mcp_tool_execution_exception("Failed to enter context manager.", inner_exception=ex) from ex async def __aexit__( self, @@ -4044,8 +4269,14 @@ def _mcp_base_span_attributes(self) -> dict[str, Any]: 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) + except Exception as ex: + _log_mcp_exception( + logging.DEBUG, + ex, + "Failed to parse URL for MCP span transport attributes", + redacted_message="Failed to parse MCP URL for span transport attributes", + sensitive_exc_info=True, + ) return attrs def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: @@ -4108,18 +4339,21 @@ async def _inject_headers(request: Request) -> None: # ruff:ignore[unused-async else: try: dynamic_headers = self._header_provider(self._connection_kwargs or {}) - except KeyError: + except KeyError as ex: # Unavoidable only when no run seeded this connection: the provider # wants per-call values a connection-lifetime request cannot have. Once # a run has seeded kwargs a missing key is a misconfiguration, and # silently dropping it would send the handshake unauthenticated. if self._connection_kwargs is not None: raise - logger.debug( + _log_mcp_exception( + logging.DEBUG, + ex, "header_provider raised KeyError for MCP server %r on an ambient " "request (no connection kwargs available); proceeding without headers.", self.name, - exc_info=True, + redacted_message="MCP header provider lacked ambient connection data", + sensitive_exc_info=True, ) dynamic_headers = {} if dynamic_headers is not None: @@ -4202,13 +4436,16 @@ async def _prepare_for_run(self, kwargs: Mapping[str, Any]) -> None: return try: headers = self._effective_headers(kwargs) - except KeyError: + except KeyError as ex: # Invocation middleware may supply runtime values unavailable during # preparation. Strict invocation-time resolution reconciles the session later. - logger.debug( + _log_mcp_exception( + logging.DEBUG, + ex, "Deferring MCP header identity reconciliation for %r until invocation.", self.name, - exc_info=True, + redacted_message="Deferred MCP header identity reconciliation after header provider failure", + sensitive_exc_info=True, ) return async with self._call_headers_lock: @@ -4266,12 +4503,12 @@ async def _ensure_session_identity( identity = _mcp_header_identity(headers) if not self._owns_session: if self._session_header_identity is None: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "MCP header identity is unknown for a caller-supplied session; " "use a separate framework-managed tool instance." ) if identity != self._session_header_identity: - raise ToolExecutionException( + raise _mcp_tool_execution_exception( "MCP header identity cannot change for a caller-supplied session; use a separate tool instance." ) return @@ -4552,8 +4789,14 @@ def _mcp_base_span_attributes(self) -> dict[str, Any]: 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) + except Exception as ex: + _log_mcp_exception( + logging.DEBUG, + ex, + "Failed to parse URL for MCP span transport attributes", + redacted_message="Failed to parse MCP URL for span transport attributes", + sensitive_exc_info=True, + ) return attrs def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]: diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index eab23d5699f..ca1e48922bc 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1269,7 +1269,10 @@ async def invoke( end_time_stamp = perf_counter() attributes[OtelAttr.ERROR_TYPE] = type(exception).__name__ capture_exception(span=span, exception=exception, timestamp=time_ns()) - logger.error(f"Function failed. Error: {exception}") + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + logger.error("Function failed. Error: %s", exception) + else: + logger.error("Function %s failed (%s).", self.name, type(exception).__name__) raise else: if skip_parsing: diff --git a/python/packages/core/agent_framework/_workflows/_events.py b/python/packages/core/agent_framework/_workflows/_events.py index 6676e9a8275..d59dc3b1a82 100644 --- a/python/packages/core/agent_framework/_workflows/_events.py +++ b/python/packages/core/agent_framework/_workflows/_events.py @@ -105,17 +105,19 @@ def from_exception( cls, exc: BaseException, *, + include_sensitive_data: bool = True, executor_id: str | None = None, extra: dict[str, Any] | None = None, ) -> WorkflowErrorDetails: tb = None - try: - tb = "".join(_traceback.format_exception(type(exc), exc, exc.__traceback__)) - except Exception: - tb = None + if include_sensitive_data: + try: + tb = "".join(_traceback.format_exception(type(exc), exc, exc.__traceback__)) + except Exception: + tb = None return cls( error_type=exc.__class__.__name__, - message=str(exc), + message=str(exc) if include_sensitive_data else exc.__class__.__name__, traceback=tb, executor_id=executor_id, extra=extra, diff --git a/python/packages/core/agent_framework/_workflows/_executor.py b/python/packages/core/agent_framework/_workflows/_executor.py index ee9734561ec..7c1a6ffd524 100644 --- a/python/packages/core/agent_framework/_workflows/_executor.py +++ b/python/packages/core/agent_framework/_workflows/_executor.py @@ -312,8 +312,14 @@ async def execute( await handler(message, context) except Exception as exc: # Surface structured executor failure before propagating + from ..observability import OBSERVABILITY_SETTINGS + + details = WorkflowErrorDetails.from_exception( + exc, + include_sensitive_data=OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED, + ) with _framework_event_origin(): - failure_event = WorkflowEvent.executor_failed(self.id, WorkflowErrorDetails.from_exception(exc)) + failure_event = WorkflowEvent.executor_failed(self.id, details) await context.add_event(failure_event) raise with _framework_event_origin(): diff --git a/python/packages/core/agent_framework/_workflows/_functional.py b/python/packages/core/agent_framework/_workflows/_functional.py index 1920b499094..0408e021ca1 100644 --- a/python/packages/core/agent_framework/_workflows/_functional.py +++ b/python/packages/core/agent_framework/_workflows/_functional.py @@ -1056,7 +1056,13 @@ async def __call__(self, *args: Any, **kwargs: Any) -> R: # from BaseException, NOT Exception, so it propagates past this handler # without emitting a spurious executor_failed event. This is intentional # — request_info is fully supported inside @step functions. - await ctx.add_event(WorkflowEvent.executor_failed(self.name, WorkflowErrorDetails.from_exception(exc))) + from ..observability import OBSERVABILITY_SETTINGS + + details = WorkflowErrorDetails.from_exception( + exc, + include_sensitive_data=OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED, + ) + await ctx.add_event(WorkflowEvent.executor_failed(self.name, details)) raise ctx._set_cached_step_auto_request_info_count( cache_key, @@ -1696,16 +1702,23 @@ async def _on_step_completed() -> None: for event in ctx._get_events(): yield event - details = WorkflowErrorDetails.from_exception(exc) + from ..observability import OBSERVABILITY_SETTINGS + + details = WorkflowErrorDetails.from_exception( + exc, + include_sensitive_data=OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED, + ) yield _framework_event(WorkflowEvent.failed, details) yield _framework_event(WorkflowEvent.status, WorkflowRunState.FAILED) + error_attributes = { + "error.type": type(exc).__name__, + } + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + error_attributes["error.message"] = str(exc) span.add_event( name=OtelAttr.WORKFLOW_ERROR, - attributes={ - "error.message": str(exc), - "error.type": type(exc).__name__, - }, + attributes=error_attributes, ) capture_exception(span, exception=exc) raise diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index d68d6f06f95..c60617dabd5 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -682,7 +682,12 @@ async def _run_workflow_with_tracing( yield event # Surface structured failure details before propagating exception - details = WorkflowErrorDetails.from_exception(exc) + from ..observability import OBSERVABILITY_SETTINGS + + details = WorkflowErrorDetails.from_exception( + exc, + include_sensitive_data=OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED, + ) with _framework_event_origin(): failed_event = WorkflowEvent.failed(details) yield failed_event @@ -690,12 +695,14 @@ async def _run_workflow_with_tracing( with _framework_event_origin(): failed_status = WorkflowEvent.status(WorkflowRunState.FAILED) yield failed_status + error_attributes = { + "error.type": type(exc).__name__, + } + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + error_attributes["error.message"] = str(exc) span.add_event( name=OtelAttr.WORKFLOW_ERROR, - attributes={ - "error.message": str(exc), - "error.type": type(exc).__name__, - }, + attributes=error_attributes, ) capture_exception(span, exception=exc) raise diff --git a/python/packages/core/agent_framework/_workflows/_workflow_builder.py b/python/packages/core/agent_framework/_workflows/_workflow_builder.py index 5d2584c1d8a..1aa75a3405d 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_builder.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_builder.py @@ -8,7 +8,7 @@ from .._agents import SupportsAgentRun from .._telemetry import FeatureIndex, mark_feature_used -from ..observability import OtelAttr, capture_exception, create_workflow_span +from ..observability import OtelAttr, create_workflow_span from ._agent_executor import AgentExecutor from ._agent_utils import resolve_agent_id from ._checkpoint import CheckpointStorage @@ -881,10 +881,12 @@ async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None: return workflow except Exception as exc: + from ..observability import OBSERVABILITY_SETTINGS + attributes = { - OtelAttr.BUILD_ERROR_MESSAGE: str(exc), OtelAttr.BUILD_ERROR_TYPE: type(exc).__name__, } + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + attributes[OtelAttr.BUILD_ERROR_MESSAGE] = str(exc) span.add_event(OtelAttr.BUILD_ERROR, attributes) # type: ignore[reportArgumentType, arg-type] - capture_exception(span, exc) raise diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 0290b916eec..fe24a7d138d 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -1006,8 +1006,8 @@ class ObservabilitySettings: metadata (e.g. token counts, operation names, durations) and does not include message content. Enabling ``enable_sensitive_data`` (env var ``ENABLE_SENSITIVE_DATA``) is an explicit, separate opt-in that additionally emits raw chat message content, function-call - arguments, and function-call results — treat that data as sensitive and ensure it is not sent - to, or retained by, a telemetry backend you have not secured appropriately. + arguments, function-call results, and exception messages — treat that data as sensitive and + ensure it is not sent to, or retained by, a telemetry backend you have not secured appropriately. Keyword Args: enable_instrumentation: Enable OpenTelemetry diagnostics. Default is True. @@ -2734,6 +2734,8 @@ def create_mcp_client_span( method_name: str, target: str | None = None, attributes: dict[str, Any] | None = None, + *, + capture_unhandled_exceptions: bool = True, ) -> Generator[trace.Span, Any, Any]: """Create an MCP client span per OTel MCP semantic conventions. @@ -2746,6 +2748,9 @@ def create_mcp_client_span( method_name: The MCP method name (e.g. ``initialize``, ``tools/call``). target: Optional low-cardinality target (tool name, prompt name). attributes: Additional span attributes. + capture_unhandled_exceptions: Whether exceptions escaping the context should be + recorded using the framework's sensitive-data policy. Internal MCP operations + that set protocol-specific error types disable this fallback. """ span_name = f"{method_name} {target}" if target else method_name attrs: dict[str, Any] = {OtelAttr.MCP_METHOD_NAME: method_name} @@ -2756,10 +2761,17 @@ def create_mcp_client_span( with trace.use_span( span=span, end_on_exit=True, - record_exception=True, - set_status_on_exception=True, + record_exception=False, + set_status_on_exception=False, ) as current_span: - yield current_span + try: + yield current_span + except Exception as exception: + if capture_unhandled_exceptions: + capture_exception(current_span, exception) + elif OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + current_span.record_exception(exception) + raise def set_mcp_span_error( @@ -2775,7 +2787,10 @@ def set_mcp_span_error( 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) + span.set_status( + trace.StatusCode.ERROR, + description=description if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED else None, + ) # endregion @@ -3191,8 +3206,11 @@ def _get_span_attributes(**kwargs: Any) -> dict[str, Any]: def capture_exception(span: trace.Span, exception: Exception, timestamp: int | None = None) -> None: """Set an error for spans.""" span.set_attribute(OtelAttr.ERROR_TYPE, type(exception).__name__) - span.record_exception(exception=exception, timestamp=timestamp) - span.set_status(status=trace.StatusCode.ERROR, description=repr(exception)) + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + span.record_exception(exception=exception, timestamp=timestamp) + span.set_status(status=trace.StatusCode.ERROR, description=repr(exception)) + else: + span.set_status(status=trace.StatusCode.ERROR) def _capture_system_instructions_latest_experimental( @@ -3646,11 +3664,24 @@ def _workflow_span_attributes( return span_attributes or None +@contextlib.contextmanager +def _capture_span_exceptions( + context_manager: contextlib.AbstractContextManager[trace.Span], +) -> Generator[trace.Span]: + """Apply the framework's sensitive-data policy to an OpenTelemetry span context.""" + with context_manager as span: + try: + yield span + except Exception as exception: + capture_exception(span, exception) + raise + + def create_workflow_span( name: str, attributes: Mapping[str, str | int] | None = None, kind: trace.SpanKind = trace.SpanKind.INTERNAL, -) -> _AgnosticContextManager[trace.Span]: +) -> contextlib.AbstractContextManager[trace.Span]: """Create a generic workflow span attached as the current span. Do not use this from an async generator that yields to callers: attaching @@ -3658,8 +3689,14 @@ def create_workflow_span( generator is later closed on GC from a different ``Context``. Streaming workflow runs should use :func:`start_workflow_span` instead. """ - return workflow_tracer().start_as_current_span( - name, kind=kind, attributes=_workflow_span_attributes(name, attributes) + return _capture_span_exceptions( + workflow_tracer().start_as_current_span( + name, + kind=kind, + attributes=_workflow_span_attributes(name, attributes), + record_exception=False, + set_status_on_exception=False, + ) ) @@ -3686,7 +3723,7 @@ def create_processing_span( payload_type: str, source_trace_contexts: list[dict[str, str]] | None = None, source_span_ids: list[str] | None = None, -) -> _AgnosticContextManager[trace.Span]: +) -> contextlib.AbstractContextManager[trace.Span]: """Create an executor processing span with optional links to source spans. Processing spans are created as children of the current workflow span and @@ -3726,16 +3763,20 @@ def create_processing_span( ) links.append(trace.Link(span_context)) - return workflow_tracer().start_as_current_span( - f"{OtelAttr.EXECUTOR_PROCESS_SPAN} {executor_id}", - kind=trace.SpanKind.INTERNAL, - attributes={ - OtelAttr.EXECUTOR_ID: executor_id, - OtelAttr.EXECUTOR_TYPE: executor_type, - OtelAttr.MESSAGE_TYPE: message_type, - OtelAttr.MESSAGE_PAYLOAD_TYPE: payload_type, - }, - links=links, + return _capture_span_exceptions( + workflow_tracer().start_as_current_span( + f"{OtelAttr.EXECUTOR_PROCESS_SPAN} {executor_id}", + kind=trace.SpanKind.INTERNAL, + attributes={ + OtelAttr.EXECUTOR_ID: executor_id, + OtelAttr.EXECUTOR_TYPE: executor_type, + OtelAttr.MESSAGE_TYPE: message_type, + OtelAttr.MESSAGE_PAYLOAD_TYPE: payload_type, + }, + links=links, + record_exception=False, + set_status_on_exception=False, + ) ) @@ -3746,7 +3787,7 @@ def create_edge_group_processing_span( message_target_id: str | None = None, source_trace_contexts: list[dict[str, str]] | None = None, source_span_ids: list[str] | None = None, -) -> _AgnosticContextManager[trace.Span]: +) -> contextlib.AbstractContextManager[trace.Span]: """Create an edge group processing span with optional links to source spans. Edge group processing spans track the processing operations in edge runners @@ -3799,9 +3840,13 @@ def create_edge_group_processing_span( # If linking fails, continue without link (graceful degradation) pass - return workflow_tracer().start_as_current_span( - f"{OtelAttr.EDGE_GROUP_PROCESS_SPAN} {edge_group_type}", - kind=trace.SpanKind.INTERNAL, - attributes=attributes, - links=links, + return _capture_span_exceptions( + workflow_tracer().start_as_current_span( + f"{OtelAttr.EDGE_GROUP_PROCESS_SPAN} {edge_group_type}", + kind=trace.SpanKind.INTERNAL, + attributes=attributes, + links=links, + record_exception=False, + set_status_on_exception=False, + ) ) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 47ca0cde917..43864b6cd1d 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -268,10 +268,13 @@ async def list_tools(params: types.PaginatedRequestParams | None = None) -> type ids=["allowed", "always_require", "never_require"], ) @pytest.mark.parametrize("remote_names", [("search", "docs_search"), ("docs_search", "search")]) +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) async def test_ambiguous_policy_reload_preserves_previous_discovery( allowed_tools: list[str] | None, approval_mode: MCPSpecificApproval | None, remote_names: tuple[str, str], + span_exporter: Any, + enable_sensitive_data: bool, caplog: pytest.LogCaptureFixture, ) -> None: tool = MCPTool( # type: ignore[abstract] # ty: ignore[call-non-callable] @@ -304,7 +307,10 @@ async def list_tools(params: types.PaginatedRequestParams | None = None) -> type await tool.message_handler(types.ServerNotification(types.ToolListChangedNotification())) await asyncio.gather(*tool._pending_reload_tasks) - assert "configuration name 'docs_search' is ambiguous" in caplog.text + private_error_detail = "configuration name 'docs_search' is ambiguous" + assert (private_error_detail in caplog.text) is enable_sensitive_data + if not enable_sensitive_data: + assert "Background MCP reload failed (ToolExecutionException)." in caplog.text assert tool._functions == original_functions assert tool.functions == original_functions assert tool._tool_call_meta_by_name == original_meta @@ -3882,10 +3888,18 @@ async def slow_load_tools(): assert len(tool._pending_reload_tasks) == 0 -async def test_mcp_tool_message_handler_reload_failure_is_logged(caplog: pytest.LogCaptureFixture): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_tool_message_handler_reload_failure_is_logged( + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """Background reload errors are logged, not raised into the receive loop.""" + private_error_detail = "founder-secret-token" tool = MCPStdioTool(name="test_tool", command="python") - tool.load_tools = AsyncMock(side_effect=RuntimeError("connection lost")) # type: ignore[method-assign] + tool.load_tools = AsyncMock( # type: ignore[method-assign] + side_effect=RuntimeError(f"connection lost: {private_error_detail}") + ) tools_notification = Mock(spec=types.ServerNotification) tools_notification.root = Mock() @@ -3900,11 +3914,14 @@ async def test_mcp_tool_message_handler_reload_failure_is_logged(caplog: pytest. tool.load_tools.assert_called_once() assert len(tool._pending_reload_tasks) == 0 - # Verify the warning was actually logged with exception info. + # Verify the warning was logged with exception info only after opt-in. reload_warnings = [r for r in caplog.records if "Background MCP reload failed" in r.message] assert len(reload_warnings) == 1 assert reload_warnings[0].levelname == "WARNING" - assert reload_warnings[0].exc_info is not None + assert (private_error_detail in caplog.text) is enable_sensitive_data + assert bool(reload_warnings[0].exc_info) is enable_sensitive_data + if not enable_sensitive_data: + assert reload_warnings[0].message == "Background MCP reload failed (RuntimeError)." async def test_mcp_tool_message_handler_cancel_and_replace(): @@ -4797,8 +4814,14 @@ async def test_connect_session_creation_failure(): assert "Session creation failed" in str(exc_info.value.__cause__) -async def test_connect_initialization_failure_http_no_command(): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_connect_initialization_failure_http_no_command( + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """Test connect() when session.initialize() fails for HTTP tool (no command attribute).""" + private_error_detail = "founder-secret-token" tool = MCPStreamableHTTPTool(name="test", url="http://example.com") # Mock successful transport creation @@ -4810,18 +4833,22 @@ async def test_connect_initialization_failure_http_no_command(): # Mock successful session creation but failed initialization mock_session = Mock() - mock_session.initialize = AsyncMock(side_effect=ConnectionError("Server not ready")) + mock_session.initialize = AsyncMock(side_effect=ConnectionError(f"Server not ready: {private_error_detail}")) with patch("mcp.client.session.ClientSession") as mock_session_class: mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) - with pytest.raises(ToolException) as exc_info: + with ( + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolException) as exc_info, + ): await tool.connect() # Should use generic error message since HTTP tool doesn't have command assert "MCP server failed to initialize" in str(exc_info.value) - assert "Server not ready" in str(exc_info.value) + assert private_error_detail in str(exc_info.value) + assert (private_error_detail in caplog.text) is enable_sensitive_data async def test_connect_cleanup_on_transport_failure(): @@ -5227,8 +5254,14 @@ async def test_connect_bare_cancel_names_cleanup_error_from_exit_stack(): assert "Cancelled via cancel scope" not in message -async def test_connect_cancelled_error_during_session_creation_logs_with_exc_info(): - """Test that CancelledError from session creation is logged with exc_info=True.""" +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_connect_cancelled_error_during_session_creation_log_respects_sensitive_data( + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that CancelledError session logs export details only after opt-in.""" + private_error_detail = "founder-secret-token" tool = MCPStreamableHTTPTool(name="test", url="http://example.com") mock_transport = (Mock(), Mock()) @@ -5238,21 +5271,17 @@ async def test_connect_cancelled_error_during_session_creation_logs_with_exc_inf tool.get_mcp_client = Mock(return_value=mock_context_manager) # type: ignore[method-assign] with patch("mcp.client.session.ClientSession") as mock_session_class: - mock_session_class.return_value.__aenter__ = AsyncMock(side_effect=asyncio.CancelledError("cancel scope")) + mock_session_class.return_value.__aenter__ = AsyncMock( + side_effect=asyncio.CancelledError(f"cancel scope: {private_error_detail}") + ) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) - from agent_framework._mcp import logger as mcp_logger - - with patch.object(mcp_logger, "debug") as mock_debug: - with pytest.raises(ToolException): - await tool.connect() + with caplog.at_level(logging.DEBUG), pytest.raises(ToolException): + await tool.connect() - # Verify logger.debug was called with exc_info=True (not an exception instance) - debug_calls = mock_debug.call_args_list - cancel_calls = [c for c in debug_calls if "Failed to create MCP session" in str(c)] - assert cancel_calls, "Expected a debug log for the cancelled session creation" - _, kwargs = cancel_calls[0] - assert kwargs.get("exc_info") is True + assert (private_error_detail in caplog.text) is enable_sensitive_data + if not enable_sensitive_data: + assert "MCP session creation failed (CancelledError)." in caplog.text def test_mcp_stdio_tool_get_mcp_client_with_env_and_kwargs(): @@ -6704,8 +6733,14 @@ async def test_connect_does_not_set_logging_level_when_logger_level_is_notset(): mock_session.set_logging_level.assert_not_called() -async def test_connect_handles_set_logging_level_exception(): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_connect_handles_set_logging_level_exception( + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """Test that connect() handles exceptions from set_logging_level gracefully.""" + private_error_detail = "founder-secret-token" tool = MCPStdioTool( name="test_server", @@ -6725,7 +6760,9 @@ async def test_connect_handles_set_logging_level_exception(): mock_session._request_id = 1 mock_session.initialize = AsyncMock() # Make set_logging_level raise an exception - mock_session.set_logging_level = AsyncMock(side_effect=RuntimeError("Server doesn't support logging level")) + mock_session.set_logging_level = AsyncMock( + side_effect=RuntimeError(f"Server doesn't support logging level: {private_error_detail}") + ) mock_session_context = AsyncMock() mock_session_context.__aenter__ = AsyncMock(return_value=mock_session) @@ -6735,7 +6772,6 @@ async def test_connect_handles_set_logging_level_exception(): patch.object(tool, "get_mcp_client", return_value=mock_context), patch("mcp.client.session.ClientSession", return_value=mock_session_context), patch.object(logger, "level", logging.INFO), # Set logger level to INFO - patch.object(logger, "warning") as mock_warning, ): # Should NOT raise - the exception should be caught and logged await tool.connect() @@ -6743,10 +6779,11 @@ async def test_connect_handles_set_logging_level_exception(): # Verify set_logging_level was called mock_session.set_logging_level.assert_called_once_with("info") - # Verify warning was logged - mock_warning.assert_called_once() - call_args = mock_warning.call_args - assert "Failed to set log level" in call_args[0][0] + assert (private_error_detail in caplog.text) is enable_sensitive_data + if enable_sensitive_data: + assert "Failed to set log level" in caplog.text + else: + assert "Failed to set MCP server log level (RuntimeError)." in caplog.text async def test_connect_reinitializes_existing_session_and_loads_tools_and_prompts() -> None: @@ -9424,9 +9461,13 @@ async def fake_connect(reset: bool = False) -> None: assert fetch_calls == 2 -async def test_fetch_task_result_second_disconnect_raises_task_state_unknown_and_cancels() -> None: - from anyio import ClosedResourceError - +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_fetch_task_result_second_disconnect_raises_task_state_unknown_and_cancels( + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: + private_error_detail = "founder-secret-token" tool = _make_task_tool() cancel_called = False @@ -9439,7 +9480,12 @@ async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> An if method == "tasks/get": return _make_task_snapshot(task_id="r2", status="completed") if method == "tasks/result": - raise ClosedResourceError + raise McpError( + types.ErrorData( + code=types.INTERNAL_ERROR, + message=f"session terminated: {private_error_detail}", + ) + ) if method == "tasks/cancel": cancel_called = True return types.CancelTaskResult() # type: ignore[call-arg] # pyrefly: ignore[missing-argument] # ty: ignore[missing-argument] @@ -9449,10 +9495,17 @@ async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> An with ( patch.object(MCPTool, "connect", new=AsyncMock(return_value=None)), - pytest.raises(ToolExecutionException, match="task state unknown"), + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolExecutionException, match="task state unknown") as exc_info, ): await tool.call_tool("slow_op") + from agent_framework import _mcp as _mcp_module + + assert isinstance(exc_info.value, _mcp_module._MCPTaskAbandoned) + assert private_error_detail in str(exc_info.value.args[1]) + assert (private_error_detail in caplog.text) is enable_sensitive_data + # Drain the fire-and-forget cancel so the assertion is deterministic. pending = list(tool._pending_reload_tasks) if pending: @@ -9593,13 +9646,18 @@ async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> An assert cancel_called is False +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) async def test_call_tool_as_task_poll_hard_mcperror_cancels_and_raises( monkeypatch: pytest.MonkeyPatch, + span_exporter: Any, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, ) -> None: from agent_framework import _mcp as _mcp_module monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1)) + private_error_detail = "founder-secret-token" tool = _make_task_tool() cancel_called = False @@ -9610,7 +9668,12 @@ async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> An if method == "tools/call": return _make_create_task_result(task_id="h1") if method == "tasks/get": - raise McpError(types.ErrorData(code=types.INVALID_PARAMS, message="bad task id")) + raise McpError( + types.ErrorData( + code=types.INVALID_PARAMS, + message=f"bad task id: {private_error_detail}", + ) + ) if method == "tasks/cancel": cancel_called = True return types.CancelTaskResult() # type: ignore[call-arg] # pyrefly: ignore[missing-argument] # ty: ignore[missing-argument] @@ -9618,9 +9681,16 @@ async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> An tool.session.send_request = AsyncMock(side_effect=fake_send) # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] - with pytest.raises(ToolExecutionException, match="bad task id"): + with ( + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolExecutionException, match="bad task id") as exc_info, + ): await tool.call_tool("slow_op") + assert isinstance(exc_info.value, _mcp_module._MCPTaskAbandoned) + assert private_error_detail in str(exc_info.value) + assert (private_error_detail in caplog.text) is enable_sensitive_data + pending = list(tool._pending_reload_tasks) if pending: await asyncio.gather(*pending, return_exceptions=True) diff --git a/python/packages/core/tests/core/test_mcp_observability.py b/python/packages/core/tests/core/test_mcp_observability.py index 7afd1dd402d..24168b1b138 100644 --- a/python/packages/core/tests/core/test_mcp_observability.py +++ b/python/packages/core/tests/core/test_mcp_observability.py @@ -7,6 +7,7 @@ from __future__ import annotations +import logging from typing import Any from unittest.mock import AsyncMock, Mock @@ -20,7 +21,7 @@ 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 +from agent_framework.observability import OtelAttr, create_mcp_client_span # region helpers @@ -93,6 +94,42 @@ def _make_get_prompt_result(text: str = "prompt result") -> types.GetPromptResul ) +def _assert_error_detail_policy(span: Any, private_error_detail: str, enable_sensitive_data: bool) -> None: + assert span.status.status_code == StatusCode.ERROR + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + assert private_error_detail in str(exception_events[0].attributes.get("exception.message")) + + +# endregion + + +# region client span error policy + + +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +def test_mcp_client_span_unhandled_exception_respects_sensitive_data( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: + """External MCP span callers should retain error classification without leaking details.""" + private_error_detail = "founder-secret-token" + + span_exporter.clear() + with pytest.raises(RuntimeError, match="MCP operation failed"): # noqa: SIM117 + # Keep these contexts nested so static analysis recognizes the following assertions as reachable. + with create_mcp_client_span("test/operation"): + raise RuntimeError(f"MCP operation failed: {private_error_detail}") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes is not None + assert span.attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + _assert_error_detail_policy(span, private_error_detail, enable_sensitive_data) + + # endregion @@ -222,38 +259,116 @@ async def test_mcp_tools_call_creates_client_span_when_no_parent(span_exporter: assert span.attributes[OtelAttr.TOOL_NAME] == "get-weather" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] -async def test_mcp_tools_call_tool_error_sets_error_type(span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_tools_call_tool_error_sets_error_type( + span_exporter: InMemorySpanExporter, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """When CallToolResult.isError is true, error.type should be 'tool_error' per MCP spec.""" + private_error_detail = "founder-secret-token" tool = _make_connected_mcp_tool() - tool.session.call_tool = AsyncMock(return_value=_make_call_tool_result("bad input", is_error=True)) # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + tool.session.call_tool = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + return_value=_make_call_tool_result(f"bad input: {private_error_detail}", is_error=True) + ) span_exporter.clear() - with pytest.raises(ToolExecutionException): + with ( + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolExecutionException) as exc_info, + ): await tool.call_tool("get-weather", city="invalid") + assert private_error_detail in str(exc_info.value) + assert (private_error_detail in caplog.text) is enable_sensitive_data 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" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] - assert span.status.status_code == StatusCode.ERROR + _assert_error_detail_policy(span, private_error_detail, enable_sensitive_data) -async def test_mcp_tools_call_mcp_error_sets_error_type(span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_tools_call_mcp_error_sets_error_type( + span_exporter: InMemorySpanExporter, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """When session.call_tool() raises McpError, error.type should be the exception class name.""" + private_error_detail = "founder-secret-token" tool = _make_connected_mcp_tool() - tool.session.call_tool = AsyncMock(side_effect=McpError(ErrorData(code=-32600, message="invalid request"))) # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + tool.session.call_tool = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + side_effect=McpError(ErrorData(code=-32600, message=f"invalid request: {private_error_detail}")) + ) span_exporter.clear() - with pytest.raises(ToolExecutionException): + with ( + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolExecutionException) as exc_info, + ): await tool.call_tool("get-weather") + assert private_error_detail in str(exc_info.value) + assert (private_error_detail in caplog.text) is enable_sensitive_data 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) == "McpError" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] - assert span.status.status_code == StatusCode.ERROR + _assert_error_detail_policy(span, private_error_detail, enable_sensitive_data) + + +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_tools_call_second_connection_loss_log_respects_sensitive_data( + span_exporter: InMemorySpanExporter, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: + """Second connection loss should retain an error log without exporting exception details by default.""" + private_error_detail = "founder-secret-token" + tool = _make_connected_mcp_tool() + tool.session.call_tool = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + side_effect=[ + McpError(ErrorData(code=-32000, message=f"session terminated: {private_error_detail}")), + McpError(ErrorData(code=-32000, message=f"session terminated: {private_error_detail}")), + ] + ) + tool.connect = AsyncMock(return_value=None) # type: ignore[method-assign] + + span_exporter.clear() + with ( + caplog.at_level(logging.ERROR, logger="agent_framework._mcp"), + pytest.raises(ToolExecutionException, match="connection lost"), + ): + await tool.call_tool("get-weather") + + assert (private_error_detail in caplog.text) is enable_sensitive_data + if not enable_sensitive_data: + assert "MCP tool call connection closed after reconnection (McpError)." in caplog.text + + +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_tools_list_error_respects_sensitive_data( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: + """tools/list errors should keep their type without exporting private details by default.""" + private_error_detail = "founder-secret-token" + tool = _make_connected_mcp_tool() + tool.session.list_tools = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + side_effect=RuntimeError(f"list failed: {private_error_detail}") + ) + + span_exporter.clear() + with pytest.raises(RuntimeError, match="list failed"): + await tool.load_tools() + + spans = span_exporter.get_finished_spans() + list_spans = [span for span in spans if span.name == "tools/list"] + assert len(list_spans) == 1 + span = list_spans[0] + assert span.attributes.get(OtelAttr.ERROR_TYPE) == "RuntimeError" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] + _assert_error_detail_policy(span, private_error_detail, enable_sensitive_data) # endregion @@ -281,23 +396,34 @@ async def test_mcp_prompts_get_creates_client_span(span_exporter: InMemorySpanEx assert span.attributes[OtelAttr.PROMPT_NAME] == "analyze-code" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] -async def test_mcp_prompts_get_mcp_error_sets_error_type(span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_mcp_prompts_get_mcp_error_sets_error_type( + span_exporter: InMemorySpanExporter, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """When session.get_prompt() raises McpError, the span should have error.type and ERROR status.""" + private_error_detail = "founder-secret-token" tool = _make_connected_mcp_tool() tool.session.get_prompt = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] - side_effect=McpError(ErrorData(code=-32602, message="prompt not found")) + side_effect=McpError(ErrorData(code=-32602, message=f"prompt not found: {private_error_detail}")) ) span_exporter.clear() - with pytest.raises(ToolExecutionException): + with ( + caplog.at_level(logging.DEBUG, logger="agent_framework"), + pytest.raises(ToolExecutionException) as exc_info, + ): await tool.get_prompt("missing-prompt") + assert private_error_detail in str(exc_info.value) + assert (private_error_detail in caplog.text) is enable_sensitive_data 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.attributes.get(OtelAttr.ERROR_TYPE) == "McpError" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] - assert span.status.status_code == StatusCode.ERROR + _assert_error_detail_policy(span, private_error_detail, enable_sensitive_data) # endregion diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index e183a746e82..1403b374a98 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -958,13 +958,17 @@ class SyncSetupAgent(AgentTelemetryLayer, _SyncSetupAgent): # type: ignore ) -async def test_function_call_with_error_handling(span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_function_call_with_error_handling( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: """Test that function call errors are properly captured in telemetry.""" + private_error_detail = "founder-secret-token" # Create a function that raises an error using the decorator @tool(name="failing_function", description="A function that fails") async def failing_function(param: str) -> str: - raise ValueError("Function execution failed") + raise ValueError(f"Function execution failed: {private_error_detail}") span_exporter.clear() @@ -986,21 +990,20 @@ async def failing_function(param: str) -> str: # Verify error status was set assert span.status.status_code == StatusCode.ERROR - assert span.status.description is not None - assert "Function execution failed" in span.status.description + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data # Verify error type attribute was set assert span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" - # Verify exception event was recorded - assert len(span.events) > 0 - exception_event = next((e for e in span.events if e.name == "exception"), None) - assert exception_event is not None - assert exception_event.attributes is not None - assert exception_event.attributes["exception.type"] == "ValueError" - exception_message = exception_event.attributes["exception.message"] - assert isinstance(exception_message, str) - assert "Function execution failed" in exception_message + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_event = exception_events[0] + assert exception_event.attributes is not None + assert exception_event.attributes["exception.type"] == "ValueError" + exception_message = exception_event.attributes["exception.message"] + assert isinstance(exception_message, str) + assert private_error_detail in exception_message # region Test OTEL environment variable parsing @@ -3157,13 +3160,16 @@ def test_create_otlp_exporters_no_endpoints(): # region Test exception handling in chat client traces -@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) -async def test_chat_client_observability_exception(mock_chat_client, span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_chat_client_observability_exception( + mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: """Test that exceptions are captured in spans.""" + private_error_detail = "founder-secret-token" class FailingChatClient(mock_chat_client): async def _inner_get_response(self, *, messages, options, **kwargs): - raise ValueError("Test error") + raise ValueError(f"Test error: {private_error_detail}") client = FailingChatClient() messages = [Message(role="user", contents=["Test"])] @@ -3176,6 +3182,15 @@ async def _inner_get_response(self, *, messages, options, **kwargs): assert len(spans) == 1 span = spans[0] assert span.status.status_code == StatusCode.ERROR + assert span.attributes is not None + assert span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) @pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) @@ -3782,9 +3797,12 @@ class MockAgent(AgentTelemetryLayer, _MockAgent): # type: ignore[misc] # pyref assert mock_emit.call_count == 0 -@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) -async def test_agent_observability_with_exception(span_exporter: InMemorySpanExporter, enable_sensitive_data): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_agent_observability_with_exception( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: """Test agent instrumentation captures exceptions.""" + private_error_detail = "founder-secret-token" class _FailingAgent: AGENT_PROVIDER_NAME = "test_provider" @@ -3812,7 +3830,7 @@ def default_options(self): return self._default_options async def run(self, messages=None, *, stream: bool = False, session=None, **kwargs): - raise RuntimeError("Agent failed") + raise RuntimeError(f"Agent failed: {private_error_detail}") class FailingAgent(AgentTelemetryLayer, _FailingAgent): # type: ignore[misc] # pyrefly: ignore[inconsistent-inheritance] # ty: ignore[invalid-method-override] pass @@ -3825,7 +3843,17 @@ class FailingAgent(AgentTelemetryLayer, _FailingAgent): # type: ignore[misc] # spans = span_exporter.get_finished_spans() assert len(spans) == 1 - assert spans[0].status.status_code == StatusCode.ERROR + span = spans[0] + assert span.status.status_code == StatusCode.ERROR + assert span.attributes is not None + assert span.attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) # region Test agent streaming observability @@ -4440,7 +4468,8 @@ def test_get_token_usage_histogram(): # region Test capture_exception -def test_capture_exception(span_exporter: InMemorySpanExporter): +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +def test_capture_exception(span_exporter: InMemorySpanExporter, enable_sensitive_data: bool) -> None: """Test capture_exception adds exception info to span.""" from time import time_ns @@ -4448,18 +4477,27 @@ def test_capture_exception(span_exporter: InMemorySpanExporter): from agent_framework.observability import capture_exception, get_tracer + private_error_detail = "founder-secret-token" span_exporter.clear() tracer = get_tracer() with tracer.start_as_current_span("test_span") as span: - exception = ValueError("Test error") + exception = ValueError(f"Test error: {private_error_detail}") capture_exception(span=span, exception=exception, timestamp=time_ns()) spans = span_exporter.get_finished_spans() assert len(spans) == 1 - assert spans[0].status.status_code == StatusCode.ERROR - # Verify exception was recorded - assert len(spans[0].events) > 0 + exported_span = spans[0] + assert exported_span.status.status_code == StatusCode.ERROR + assert exported_span.attributes is not None + assert exported_span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert (private_error_detail in (exported_span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in exported_span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) # region Test _get_span diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py index c75d3ec88a5..7a7c99580f7 100644 --- a/python/packages/core/tests/core/test_tools.py +++ b/python/packages/core/tests/core/test_tools.py @@ -1167,10 +1167,14 @@ def pydantic_test_tool(x: int, y: int) -> int: assert span.attributes[OtelAttr.TOOL_ARGUMENTS] == '{"x": 5, "y": 10}' # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) async def test_tool_invoke_telemetry_with_exception( - span_exporter: InMemorySpanExporter, caplog: pytest.LogCaptureFixture -): + span_exporter: InMemorySpanExporter, + enable_sensitive_data: bool, + caplog: pytest.LogCaptureFixture, +) -> None: """Test the tool invoke method with telemetry when an exception occurs.""" + private_error_detail = "founder-secret-token" @tool( name="exception_test_tool", @@ -1178,7 +1182,7 @@ async def test_tool_invoke_telemetry_with_exception( ) def exception_test_tool(x: int, y: int) -> int: """A function that raises an exception for telemetry testing.""" - raise ValueError("Test exception for telemetry") + raise ValueError(f"Test exception for telemetry: {private_error_detail}") mock_histogram = Mock() exception_test_tool._invocation_duration_histogram = mock_histogram @@ -1186,20 +1190,35 @@ def exception_test_tool(x: int, y: int) -> int: # Call invoke and expect exception with pytest.raises(ValueError, match="Test exception for telemetry"): await exception_test_tool.invoke(x=1, y=2, tool_call_id="exception_call") - assert "Function failed. Error: Test exception for telemetry" in caplog.text + if enable_sensitive_data: + assert f"Function failed. Error: Test exception for telemetry: {private_error_detail}" in caplog.text + else: + assert "Function exception_test_tool failed (ValueError)." in caplog.text + assert private_error_detail not in caplog.text assert "result parser failed" not in caplog.text spans = span_exporter.get_finished_spans() assert len(spans) == 1 span = spans[0] assert OtelAttr.TOOL_EXECUTION_OPERATION.value in span.name assert "exception_test_tool" in span.name + assert span.attributes is not None assert span.attributes[OtelAttr.TOOL_NAME] == "exception_test_tool" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert span.attributes[OtelAttr.TOOL_CALL_ID] == "exception_call" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert span.attributes[OtelAttr.TOOL_TYPE] == "function" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert span.attributes[OtelAttr.TOOL_DESCRIPTION] == "A test tool that raises an exception" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] - assert span.attributes[OtelAttr.TOOL_ARGUMENTS] == '{"x": 1, "y": 2}' # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] + if enable_sensitive_data: + assert span.attributes[OtelAttr.TOOL_ARGUMENTS] == '{"x": 1, "y": 2}' # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] + else: + assert OtelAttr.TOOL_ARGUMENTS not in span.attributes assert span.attributes[OtelAttr.ERROR_TYPE] == ValueError.__name__ # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert span.status.status_code == trace.StatusCode.ERROR + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) # Verify histogram was called with error attributes mock_histogram.record.assert_called_once() diff --git a/python/packages/core/tests/workflow/test_edge.py b/python/packages/core/tests/workflow/test_edge.py index 2c62e245be1..e054a73e2a3 100644 --- a/python/packages/core/tests/workflow/test_edge.py +++ b/python/packages/core/tests/workflow/test_edge.py @@ -25,7 +25,7 @@ ) from agent_framework._workflows._edge_runner import create_edge_runner from agent_framework._workflows._state import State -from agent_framework.observability import EdgeGroupDeliveryStatus +from agent_framework.observability import EdgeGroupDeliveryStatus, OtelAttr @dataclass @@ -388,6 +388,45 @@ async def test_single_edge_group_tracing_condition_failure(span_exporter: InMemo assert span.attributes.get("edge_group.delivery_status") == EdgeGroupDeliveryStatus.DROPPED_CONDITION_FALSE.value +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_single_edge_group_tracing_exception_respects_sensitive_data( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: + """Edge processing spans should keep error metadata without exporting private exception details.""" + private_error_detail = "founder-secret-token" + source = MockExecutor(id="source_executor") + target = MockExecutor(id="target_executor") + + def failing_condition(_: MockMessage) -> bool: + raise RuntimeError(f"Condition failed: {private_error_detail}") + + executors: dict[str, Executor] = {source.id: source, target.id: target} + edge_group = SingleEdgeGroup(source_id=source.id, target_id=target.id, condition=failing_condition) + edge_runner = create_edge_runner(edge_group, executors) + state = State() + ctx = InProcRunnerContext() + message = WorkflowMessage(data=MockMessage(data="fail"), source_id=source.id) + + span_exporter.clear() + with pytest.raises(RuntimeError, match="Condition failed"): + await edge_runner.send_message(message, state, ctx) + + spans = span_exporter.get_finished_spans() + edge_group_spans = [span for span in spans if span.name == "edge_group.process SingleEdgeGroup"] + assert len(edge_group_spans) == 1 + span = edge_group_spans[0] + assert span.attributes is not None + assert span.attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert span.status.status_code.name == "ERROR" + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) + + async def test_single_edge_group_tracing_type_mismatch(span_exporter: InMemorySpanExporter) -> None: """Test that single edge group processing creates proper spans for type mismatches.""" source = MockExecutor(id="source_executor") diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py index daaea7682fc..6a627a5e690 100644 --- a/python/packages/core/tests/workflow/test_workflow_observability.py +++ b/python/packages/core/tests/workflow/test_workflow_observability.py @@ -1,12 +1,14 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio from typing import Any, cast import pytest from opentelemetry import trace from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder +from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder, step, workflow +from agent_framework._workflows._events import WorkflowErrorDetails, WorkflowEvent from agent_framework._workflows._executor import Executor, handler from agent_framework._workflows._runner_context import InProcRunnerContext, MessageType, WorkflowMessage from agent_framework._workflows._state import State @@ -19,6 +21,21 @@ ) +def _assert_workflow_error_details( + details: WorkflowErrorDetails, + private_error_detail: str, + enable_sensitive_data: bool, +) -> None: + assert private_error_detail not in details.error_type + if enable_sensitive_data: + assert private_error_detail in details.message + assert details.traceback is not None + assert private_error_detail in details.traceback + else: + assert details.message == details.error_type + assert details.traceback is None + + class MockExecutor(Executor): """Mock executor for testing.""" @@ -408,8 +425,12 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter) assert len(aggregator_span.links) >= 2, f"Expected at least 2 links, got {len(aggregator_span.links)}" -async def test_workflow_error_handling_in_tracing(span_exporter: InMemorySpanExporter) -> None: +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_workflow_error_handling_in_tracing( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: """Test that workflow errors are properly recorded in traces.""" + private_error_detail = "founder-secret-token" class FailingExecutor(Executor): def __init__(self) -> None: @@ -417,15 +438,27 @@ def __init__(self) -> None: @handler async def handle_message(self, message: str, ctx: WorkflowContext) -> None: - raise ValueError("Test error") + raise ValueError(f"Test error: {private_error_detail}") failing_executor = FailingExecutor() workflow = WorkflowBuilder(start_executor=failing_executor).build() + span_exporter.clear() # Run workflow and expect error + events: list[WorkflowEvent[Any]] = [] with pytest.raises(ValueError, match="Test error"): - async for _ in workflow.run("test input", stream=True): - pass + async for event in workflow.run("test input", stream=True): + events.append(event) + + failed_events = [event for event in events if event.type == "failed"] + assert len(failed_events) == 1 + assert failed_events[0].details is not None + _assert_workflow_error_details(failed_events[0].details, private_error_detail, enable_sensitive_data) + + executor_failed_events = [event for event in events if event.type == "executor_failed"] + assert len(executor_failed_events) == 1 + assert executor_failed_events[0].details is not None + _assert_workflow_error_details(executor_failed_events[0].details, private_error_detail, enable_sensitive_data) spans = span_exporter.get_finished_spans() @@ -441,6 +474,116 @@ async def handle_message(self, message: str, ctx: WorkflowContext) -> None: assert "workflow.started" in event_names assert "workflow.error" in event_names assert workflow_span.status.status_code.name == "ERROR" + assert workflow_span.attributes is not None + assert workflow_span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert (private_error_detail in (workflow_span.status.description or "")) is enable_sensitive_data + + workflow_error = next(event for event in workflow_span.events if event.name == "workflow.error") + assert workflow_error.attributes is not None + assert workflow_error.attributes["error.type"] == "ValueError" + assert ("error.message" in workflow_error.attributes) is enable_sensitive_data + if enable_sensitive_data: + assert private_error_detail in str(workflow_error.attributes["error.message"]) + + workflow_exception_events = [event for event in workflow_span.events if event.name == "exception"] + assert bool(workflow_exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = workflow_exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) + + executor_spans = [span for span in spans if span.name == "executor.process failing_executor"] + assert len(executor_spans) == 1 + executor_span = executor_spans[0] + assert executor_span.attributes is not None + assert executor_span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert executor_span.status.status_code.name == "ERROR" + assert (private_error_detail in (executor_span.status.description or "")) is enable_sensitive_data + executor_exception_events = [event for event in executor_span.events if event.name == "exception"] + assert bool(executor_exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = executor_exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) + + +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_functional_workflow_error_respects_sensitive_data( + span_exporter: InMemorySpanExporter, enable_sensitive_data: bool +) -> None: + """Functional workflow spans should redact exception details unless explicitly enabled.""" + private_error_detail = "founder-secret-token" + + @step + async def failing_step(message: str) -> None: + raise RuntimeError(f"Functional workflow failed: {message}") + + @workflow + async def failing_workflow(message: str) -> None: + await failing_step(message) + + workflow_instance = failing_workflow.build() + span_exporter.clear() + stream = workflow_instance.run(private_error_detail, stream=True) + events: list[WorkflowEvent[Any]] = [] + with pytest.raises(RuntimeError, match="Functional workflow failed"): + async for event in stream: + events.append(event) + + failed_events = [event for event in events if event.type == "failed"] + assert len(failed_events) == 1 + assert failed_events[0].details is not None + _assert_workflow_error_details(failed_events[0].details, private_error_detail, enable_sensitive_data) + + executor_failed_events = [event for event in events if event.type == "executor_failed"] + assert len(executor_failed_events) == 1 + assert executor_failed_events[0].details is not None + _assert_workflow_error_details(executor_failed_events[0].details, private_error_detail, enable_sensitive_data) + + spans = span_exporter.get_finished_spans() + workflow_spans = [span for span in spans if span.name == "workflow.run"] + assert len(workflow_spans) == 1 + span = workflow_spans[0] + assert span.attributes is not None + assert span.attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert span.status.status_code.name == "ERROR" + assert (private_error_detail in (span.status.description or "")) is enable_sensitive_data + + error_event = next(event for event in span.events if event.name == "workflow.error") + assert error_event.attributes is not None + assert error_event.attributes["error.type"] == "RuntimeError" + assert ("error.message" in error_event.attributes) is enable_sensitive_data + if enable_sensitive_data: + assert private_error_detail in str(error_event.attributes["error.message"]) + + exception_events = [event for event in span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data + if enable_sensitive_data: + exception_attributes = exception_events[0].attributes + assert exception_attributes is not None + assert private_error_detail in str(exception_attributes.get("exception.message")) + + +@pytest.mark.parametrize("exception_type", [asyncio.CancelledError, GeneratorExit]) +def test_workflow_span_lifecycle_exceptions_are_not_failures( + span_exporter: InMemorySpanExporter, + exception_type: type[BaseException], +) -> None: + """Workflow span helpers should not classify lifecycle-control exceptions as failures.""" + span_exporter.clear() + + try: + with create_workflow_span("workflow.lifecycle"): + raise exception_type() + except exception_type: + pass + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.status.status_code != trace.StatusCode.ERROR + assert not span.attributes or OtelAttr.ERROR_TYPE not in span.attributes + assert all(event.name != "exception" for event in span.events) @pytest.mark.parametrize("enable_instrumentation", [False], indirect=True) @@ -480,7 +623,8 @@ async def test_message_trace_context_serialization(span_exporter: InMemorySpanEx assert restored_msg.source_span_ids == ["span123"] # Test new format -async def test_workflow_build_error_tracing(span_exporter: InMemorySpanExporter) -> None: +@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True) +async def test_workflow_build_error_tracing(span_exporter: InMemorySpanExporter, enable_sensitive_data: bool) -> None: """Test that build errors are properly recorded in build spans.""" # Create a valid builder, then clear the start executor to trigger a build-time ValueError @@ -510,5 +654,13 @@ async def test_workflow_build_error_tracing(span_exporter: InMemorySpanExporter) error_event = error_events[0] assert error_event.attributes is not None - assert "starting executor" in str(error_event.attributes.get("build.error.message")).lower() assert error_event.attributes.get("build.error.type") == "ValueError" + assert ("build.error.message" in error_event.attributes) is enable_sensitive_data + if enable_sensitive_data: + assert "starting executor" in str(error_event.attributes["build.error.message"]).lower() + + assert build_span.attributes is not None + assert build_span.attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert ("starting executor" in (build_span.status.description or "").lower()) is enable_sensitive_data + exception_events = [event for event in build_span.events if event.name == "exception"] + assert bool(exception_events) is enable_sensitive_data