diff --git a/src/agents/mcp/server.py b/src/agents/mcp/server.py index 944d0d1738..35cf10ace8 100644 --- a/src/agents/mcp/server.py +++ b/src/agents/mcp/server.py @@ -101,6 +101,9 @@ class RequireApprovalObject(TypedDict, total=False): T = TypeVar("T") +_SAFE_EXCEPTION_GROUP_MESSAGE = "MCP request failed with additional errors." +_SAFE_EXCEPTION_MESSAGE = "An additional error occurred during the MCP request." + def _safe_transport_cause(http_error: Exception) -> Exception | None: """Keep a transport exception only when its HTTPX URLs need no sanitization.""" @@ -136,6 +139,37 @@ def _first_unsafe_transport_error(http_errors: list[Exception]) -> Exception | N return next((error for error in http_errors if _safe_transport_cause(error) is None), None) +def _is_http_transport_error(error: BaseException) -> bool: + """Return whether an exception is an HTTPX transport error.""" + return isinstance(error, httpx.HTTPStatusError | httpx.RequestError) + + +def _credential_safe_exception_group(error_group: BaseExceptionGroup) -> BaseExceptionGroup: + """Replace an exception group with a fixed-data graph that retains control semantics.""" + safe_exceptions = [ + _credential_safe_exception_group(error) + if isinstance(error, BaseExceptionGroup) + else _credential_safe_exception_leaf(error) + for error in error_group.exceptions + ] + return BaseExceptionGroup(_SAFE_EXCEPTION_GROUP_MESSAGE, safe_exceptions) + + +def _credential_safe_exception_leaf(error: BaseException) -> BaseException: + """Create a fixed-data replacement for one retained exception leaf.""" + if isinstance(error, asyncio.CancelledError): + return asyncio.CancelledError() + if isinstance(error, KeyboardInterrupt): + return KeyboardInterrupt() + if isinstance(error, SystemExit): + return SystemExit() + if isinstance(error, GeneratorExit): + return GeneratorExit() + if isinstance(error, Exception): + return RuntimeError(_SAFE_EXCEPTION_MESSAGE) + return BaseException(_SAFE_EXCEPTION_MESSAGE) + + def _log_transport_warning(message: str, http_error: Exception) -> None: """Log a transport failure without attaching credential-bearing request URLs.""" if _debug.DONT_LOG_TOOL_DATA: @@ -826,6 +860,64 @@ def _raise_mapped_transport_error(error: UserError, cause: Exception | None) -> raise error from None raise error from cause + def _user_error_for_request_operation( + self, + operation: str, + http_error: Exception, + ) -> UserError: + """Build a credential-safe error for an MCP request operation.""" + error_message = f"Failed to {operation} on MCP server '{self._error_name}': " + if isinstance(http_error, httpx.HTTPStatusError): + error_message += f"HTTP error {http_error.response.status_code}" + elif isinstance(http_error, httpx.ConnectError): + error_message += "Connection lost. The server may have disconnected." + elif isinstance(http_error, httpx.TimeoutException): + error_message += "Connection timeout." + else: + error_message += "Request failed." + return UserError(error_message) + + async def _run_request_with_transport_error_redaction( + self, + operation: str, + func: Callable[[], Awaitable[T]], + ) -> T: + """Run an MCP request without retaining credential-bearing HTTP errors.""" + transport_error: UserError | None = None + base_error_group: BaseExceptionGroup | None = None + try: + return await func() + except (httpx.HTTPStatusError, httpx.RequestError) as http_error: + transport_error = self._user_error_for_request_operation(operation, http_error) + except BaseExceptionGroup as error_group: + http_errors = self._extract_http_errors_from_exception(error_group) + if not http_errors: + raise + selected_http_error = http_errors[0] + http_group, remaining_group = error_group.split(_is_http_transport_error) + assert http_group is not None + mapped_transport_error = self._user_error_for_request_operation( + operation, + selected_http_error, + ) + if remaining_group is None: + transport_error = mapped_transport_error + else: + safe_remaining_group = _credential_safe_exception_group(remaining_group) + base_error_group = BaseExceptionGroup( + _SAFE_EXCEPTION_GROUP_MESSAGE, + [mapped_transport_error, *safe_remaining_group.exceptions], + ) + http_errors.clear() + del selected_http_error + del http_group + del remaining_group + + if base_error_group is not None: + raise base_error_group + assert transport_error is not None + self._raise_mapped_transport_error(transport_error, None) + async def _run_with_retries(self, func: Callable[[], Awaitable[T]]) -> T: attempts = 0 while True: @@ -1079,7 +1171,10 @@ async def list_prompts( raise UserError("Server not initialized. Make sure you call `connect()` first.") session = self.session assert session is not None - return await self._maybe_serialize_request(lambda: session.list_prompts()) + return await self._run_request_with_transport_error_redaction( + "list prompts", + lambda: self._maybe_serialize_request(lambda: session.list_prompts()), + ) async def get_prompt( self, name: str, arguments: dict[str, Any] | None = None @@ -1089,7 +1184,10 @@ async def get_prompt( raise UserError("Server not initialized. Make sure you call `connect()` first.") session = self.session assert session is not None - return await self._maybe_serialize_request(lambda: session.get_prompt(name, arguments)) + return await self._run_request_with_transport_error_redaction( + "get prompt", + lambda: self._maybe_serialize_request(lambda: session.get_prompt(name, arguments)), + ) async def list_resources(self, cursor: str | None = None) -> ListResourcesResult: """List the resources available on the server.""" @@ -1097,7 +1195,10 @@ async def list_resources(self, cursor: str | None = None) -> ListResourcesResult raise UserError("Server not initialized. Make sure you call `connect()` first.") session = self.session assert session is not None - return await self._maybe_serialize_request(lambda: session.list_resources(cursor)) + return await self._run_request_with_transport_error_redaction( + "list resources", + lambda: self._maybe_serialize_request(lambda: session.list_resources(cursor)), + ) async def list_resource_templates( self, cursor: str | None = None @@ -1107,7 +1208,10 @@ async def list_resource_templates( raise UserError("Server not initialized. Make sure you call `connect()` first.") session = self.session assert session is not None - return await self._maybe_serialize_request(lambda: session.list_resource_templates(cursor)) + return await self._run_request_with_transport_error_redaction( + "list resource templates", + lambda: self._maybe_serialize_request(lambda: session.list_resource_templates(cursor)), + ) async def read_resource(self, uri: str) -> ReadResourceResult: """Read the contents of a specific resource by URI. @@ -1122,7 +1226,10 @@ async def read_resource(self, uri: str) -> ReadResourceResult: assert session is not None from pydantic import AnyUrl - return await self._maybe_serialize_request(lambda: session.read_resource(AnyUrl(uri))) + return await self._run_request_with_transport_error_redaction( + "read resource", + lambda: self._maybe_serialize_request(lambda: session.read_resource(AnyUrl(uri))), + ) async def cleanup(self): """Cleanup the server.""" diff --git a/src/agents/voice/result.py b/src/agents/voice/result.py index 15b196bf95..9e5641384d 100644 --- a/src/agents/voice/result.py +++ b/src/agents/voice/result.py @@ -289,18 +289,25 @@ async def _wait_for_completion(self): tasks.append(self._dispatcher_task) await asyncio.gather(*tasks) - def _cleanup_tasks(self): - self._finish_turn() + async def _cleanup_tasks(self): + current_task = asyncio.current_task() + tasks: list[asyncio.Task[Any]] = [] + seen: set[asyncio.Task[Any]] = set() + for task in [*self._tasks, self._dispatcher_task, self.text_generation_task]: + if task is None or task is current_task or task in seen: + continue + seen.add(task) + tasks.append(task) - for task in self._tasks: + for task in tasks: if not task.done(): task.cancel() - if self._dispatcher_task and not self._dispatcher_task.done(): - self._dispatcher_task.cancel() - - if self.text_generation_task and not self.text_generation_task.done(): - self.text_generation_task.cancel() + try: + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + finally: + self._finish_turn() def _check_errors(self): for task in self._tasks: @@ -316,7 +323,7 @@ async def stream(self) -> AsyncIterator[VoiceStreamEvent]: try: event = await self._queue.get() except asyncio.CancelledError: - self._cleanup_tasks() + await self._cleanup_tasks() raise if isinstance(event, VoiceStreamEventError): self._stored_exception = event.error @@ -333,15 +340,17 @@ async def stream(self) -> AsyncIterator[VoiceStreamEvent]: # On the normal completion path, let the producer task finish gracefully so any active # trace context can emit `trace_end` before we run cleanup. - if ( - saw_session_end - and self.text_generation_task is not None - and not self.text_generation_task.done() - ): - await asyncio.shield(self.text_generation_task) - - self._check_errors() - self._cleanup_tasks() + try: + if ( + saw_session_end + and self.text_generation_task is not None + and not self.text_generation_task.done() + ): + await asyncio.shield(self.text_generation_task) + + self._check_errors() + finally: + await self._cleanup_tasks() if self._stored_exception: raise self._stored_exception diff --git a/tests/mcp/test_server_errors.py b/tests/mcp/test_server_errors.py index 5f10da4ce7..b5ffc88406 100644 --- a/tests/mcp/test_server_errors.py +++ b/tests/mcp/test_server_errors.py @@ -29,6 +29,13 @@ ) _URL_SECRETS = ("user", "s3cr3t_pw", "SECRET_QS_KEY", "SECRET_FRAGMENT") _SAFE_URL = "https://mcp.example.com/sse" +_PROMPT_RESOURCE_OPERATIONS = [ + ("list_prompts", (), "list prompts"), + ("get_prompt", ("safe_prompt", None), "get prompt"), + ("list_resources", (None,), "list resources"), + ("list_resource_templates", (None,), "list resource templates"), + ("read_resource", ("file:///safe.txt",), "read resource"), +] def _assert_url_credentials_hidden(error: BaseException) -> None: @@ -48,6 +55,47 @@ def _assert_not_retained_in_traceback_locals(error: BaseException, sensitive_val current = current.tb_next +def _assert_not_retained_in_exception_graph( + error: BaseException, + sensitive_value: object, +) -> None: + pending: list[object] = [error] + seen: set[int] = set() + + while pending: + value = pending.pop() + assert value is not sensitive_value + if id(value) in seen: + continue + seen.add(id(value)) + + if isinstance(value, BaseException): + pending.extend(value.args) + if value.__cause__ is not None: + pending.append(value.__cause__) + if value.__context__ is not None: + pending.append(value.__context__) + pending.extend(getattr(value, "__notes__", ())) + pending.append(value.__dict__) + if isinstance(value, BaseExceptionGroup): + pending.extend(value.exceptions) + elif isinstance(value, dict): + pending.extend(value.keys()) + pending.extend(value.values()) + elif isinstance(value, list | tuple | set | frozenset): + pending.extend(value) + + +def _assert_url_credentials_hidden_from_traceback_locals(error: BaseException) -> None: + current = error.__traceback__ + while current is not None: + if current.tb_frame.f_code.co_filename.endswith("/src/agents/mcp/server.py"): + attached_values = repr(tuple(current.tb_frame.f_locals.values())) + for secret in _URL_SECRETS: + assert secret not in attached_values + current = current.tb_next + + def _assert_url_credentials_hidden_from_log_record(record: logging.LogRecord) -> None: rendered = logging.Formatter("%(levelname)s %(message)s").format(record) attached_values = repr( @@ -126,6 +174,264 @@ async def test_not_calling_connect_causes_error(): await server.call_tool("foo", {}) +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method_name", "args", "operation"), + _PROMPT_RESOURCE_OPERATIONS, +) +@pytest.mark.parametrize("redacted", [True, False]) +async def test_prompt_and_resource_request_errors_hide_url_credentials( + monkeypatch, + caplog, + method_name: str, + args: tuple[object, ...], + operation: str, + redacted: bool, +): + monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", redacted) + server = MCPServerStreamableHttp(params={"url": _CREDENTIALED_URL}) + request_error = httpx.ReadError( + "request failed", + request=httpx.Request("POST", _CREDENTIALED_URL), + ) + session = MagicMock() + setattr(session, method_name, AsyncMock(side_effect=request_error)) + server.session = session + + with caplog.at_level(logging.DEBUG, logger="openai.agents"): + with pytest.raises(UserError) as user_error_info: + await getattr(server, method_name)(*args) + + assert f"Failed to {operation}" in str(user_error_info.value) + assert "mcp.example.com/sse" in str(user_error_info.value) + assert "Request failed" in str(user_error_info.value) + assert not hasattr(user_error_info.value, "request") + _assert_url_credentials_hidden(user_error_info.value) + _assert_not_retained_in_traceback_locals(user_error_info.value, request_error) + _assert_url_credentials_hidden_from_traceback_locals(user_error_info.value) + assert not [record for record in caplog.records if record.name == "openai.agents"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method_name", "args", "_operation"), + _PROMPT_RESOURCE_OPERATIONS, +) +async def test_prompt_and_resource_request_errors_hide_attached_request_data( + method_name: str, + args: tuple[object, ...], + _operation: str, +): + session_secret = "SECRET_MCP_SESSION_ID" + body_secret = "SECRET_REQUEST_BODY" + server = MCPServerStreamableHttp(params={"url": _SAFE_URL}) + request_error = httpx.ReadError( + "request failed", + request=httpx.Request( + "POST", + _SAFE_URL, + headers={"mcp-session-id": session_secret}, + content=body_secret, + ), + ) + session = MagicMock() + setattr(session, method_name, AsyncMock(side_effect=request_error)) + server.session = session + + with pytest.raises(UserError) as user_error_info: + await getattr(server, method_name)(*args) + + rendered = "".join(traceback.format_exception(user_error_info.value)) + assert session_secret not in rendered + assert body_secret not in rendered + assert user_error_info.value.__cause__ is None + assert user_error_info.value.__context__ is None + _assert_not_retained_in_traceback_locals(user_error_info.value, request_error) + _assert_not_retained_in_exception_graph(user_error_info.value, request_error) + + +@pytest.mark.asyncio +async def test_prompt_http_status_errors_hide_attached_response_data(): + request_body_secret = "SECRET_REQUEST_BODY" + response_header_secret = "SECRET_RESPONSE_COOKIE" + response_body_secret = "SECRET_RESPONSE_BODY" + history_body_secret = "SECRET_HISTORY_BODY" + server = MCPServerStreamableHttp(params={"url": _SAFE_URL}) + request = httpx.Request("POST", _SAFE_URL, content=request_body_secret) + history_request = httpx.Request("POST", _SAFE_URL) + history_response = httpx.Response( + 307, + request=history_request, + headers={"set-cookie": history_body_secret}, + content=history_body_secret, + ) + response = httpx.Response( + 503, + request=request, + headers={"set-cookie": response_header_secret}, + content=response_body_secret, + history=[history_response], + ) + http_error = httpx.HTTPStatusError("boom", request=request, response=response) + session = MagicMock() + session.list_prompts = AsyncMock(side_effect=http_error) + server.session = session + + with pytest.raises(UserError) as user_error_info: + await server.list_prompts() + + rendered = "".join(traceback.format_exception(user_error_info.value)) + for secret in ( + request_body_secret, + response_header_secret, + response_body_secret, + history_body_secret, + ): + assert secret not in rendered + assert user_error_info.value.__cause__ is None + assert user_error_info.value.__context__ is None + _assert_not_retained_in_traceback_locals(user_error_info.value, http_error) + _assert_not_retained_in_exception_graph(user_error_info.value, http_error) + + +@pytest.mark.asyncio +async def test_prompt_request_http_status_hides_url_credentials(): + server = MCPServerStreamableHttp(params={"url": _CREDENTIALED_URL}) + request = httpx.Request("GET", _CREDENTIALED_URL) + http_error = httpx.HTTPStatusError( + "boom", + request=request, + response=httpx.Response(503, request=request), + ) + session = MagicMock() + session.list_prompts = AsyncMock(side_effect=http_error) + server.session = session + + with pytest.raises(UserError) as user_error_info: + await server.list_prompts() + + assert "HTTP error 503" in str(user_error_info.value) + _assert_url_credentials_hidden(user_error_info.value) + _assert_not_retained_in_traceback_locals(user_error_info.value, http_error) + _assert_url_credentials_hidden_from_traceback_locals(user_error_info.value) + + +@pytest.mark.asyncio +async def test_resource_request_nested_group_replaces_ordinary_siblings_safely(): + server = MCPServerStreamableHttp(params={"url": _CREDENTIALED_URL}) + request_error = httpx.ConnectError( + "connection failed", + request=httpx.Request("GET", _CREDENTIALED_URL), + ) + + ordinary_error = ValueError("ordinary sibling failure", request_error) + ordinary_error.__notes__ = [_CREDENTIALED_URL] + ordinary_error.unsafe_request = request_error # type: ignore[attr-defined] + error_group = BaseExceptionGroup( + "request failed", + [ + ordinary_error, + BaseExceptionGroup("transport failed", [request_error]), + ], + ) + session = MagicMock() + session.read_resource = AsyncMock(side_effect=error_group) + server.session = session + + with pytest.raises(BaseExceptionGroup) as error_group_info: + await server.read_resource("file:///safe.txt") + + propagated_group = error_group_info.value + assert len(propagated_group.exceptions) == 2 + propagated_transport_error, propagated_error = propagated_group.exceptions + assert isinstance(propagated_transport_error, UserError) + assert "Failed to read resource" in str(propagated_transport_error) + assert "Connection lost" in str(propagated_transport_error) + assert propagated_transport_error.__cause__ is None + assert propagated_transport_error.__context__ is None + assert isinstance(propagated_error, RuntimeError) + assert str(propagated_error) == "An additional error occurred during the MCP request." + assert id(propagated_error) != id(ordinary_error) + _assert_url_credentials_hidden(propagated_group) + _assert_not_retained_in_traceback_locals(propagated_group, error_group) + _assert_not_retained_in_traceback_locals(propagated_group, request_error) + _assert_not_retained_in_exception_graph(propagated_group, ordinary_error) + _assert_not_retained_in_exception_graph(propagated_group, request_error) + _assert_url_credentials_hidden_from_traceback_locals(propagated_group) + + +@pytest.mark.asyncio +async def test_resource_request_mixed_group_preserves_cancellation(): + server = MCPServerStreamableHttp(params={"url": _CREDENTIALED_URL}) + cancellation = asyncio.CancelledError("request cancelled") + request_error = httpx.ConnectError( + "connection failed", + request=httpx.Request("GET", _CREDENTIALED_URL), + ) + error_group: BaseExceptionGroup | None = None + + async def raise_mixed_group(uri: object) -> None: + del uri + nonlocal error_group + error_group = BaseExceptionGroup( + "request failed", + [cancellation, request_error], + ) + raise error_group + + session = MagicMock() + session.read_resource = raise_mixed_group + server.session = session + + with pytest.raises(BaseExceptionGroup) as error_group_info: + await server.read_resource("file:///safe.txt") + + propagated_group = error_group_info.value + assert len(propagated_group.exceptions) == 2 + propagated_transport_error, propagated_cancellation = propagated_group.exceptions + assert isinstance(propagated_transport_error, UserError) + assert "Failed to read resource" in str(propagated_transport_error) + assert "Connection lost" in str(propagated_transport_error) + assert propagated_transport_error.__cause__ is None + assert propagated_transport_error.__context__ is None + assert isinstance(propagated_cancellation, asyncio.CancelledError) + assert propagated_cancellation is not cancellation + _assert_url_credentials_hidden(propagated_group) + assert error_group is not None + _assert_not_retained_in_traceback_locals(propagated_group, error_group) + _assert_not_retained_in_traceback_locals(propagated_group, request_error) + _assert_not_retained_in_exception_graph(propagated_group, cancellation) + _assert_not_retained_in_exception_graph(propagated_group, request_error) + _assert_url_credentials_hidden_from_traceback_locals(propagated_group) + traceback_frames = [] + current = propagated_group.__traceback__ + while current is not None: + traceback_frames.append(current.tb_frame) + current = current.tb_next + assert all(frame.f_code.co_name != "raise_mixed_group" for frame in traceback_frames) + + +@pytest.mark.asyncio +async def test_resource_request_sanitizes_safe_url_nested_group(): + server = MCPServerStreamableHttp(params={"url": _SAFE_URL}) + request_error = httpx.ConnectError( + "connection failed", + request=httpx.Request("GET", _SAFE_URL), + ) + error_group = BaseExceptionGroup("request failed", [request_error]) + session = MagicMock() + session.read_resource = AsyncMock(side_effect=error_group) + server.session = session + + with pytest.raises(UserError) as user_error_info: + await server.read_resource("file:///safe.txt") + + assert user_error_info.value.__cause__ is None + assert user_error_info.value.__context__ is None + _assert_not_retained_in_traceback_locals(user_error_info.value, error_group) + _assert_not_retained_in_exception_graph(user_error_info.value, request_error) + + @pytest.mark.asyncio @pytest.mark.parametrize( ("url", "retains_cause"), diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py index 0163ca7999..b76b51555a 100644 --- a/tests/voice/test_pipeline.py +++ b/tests/voice/test_pipeline.py @@ -288,6 +288,51 @@ async def run(self, text: str, settings: TTSModelSettings): assert len(terminal_events) == 1 +@pytest.mark.asyncio +async def test_voice_pipeline_awaits_task_cleanup_after_tts_failure() -> None: + """A public pipeline stream must await sibling task cleanup when TTS fails.""" + + second_segment_started = asyncio.Event() + second_segment_stopped = asyncio.Event() + + class FailingTTS(FakeTTS): + async def run(self, text: str, settings: TTSModelSettings): + del settings + if text == "first": + await second_segment_started.wait() + raise RuntimeError("tts-failure") + yield b"" # pragma: no cover + + second_segment_started.set() + try: + await asyncio.Event().wait() + finally: + second_segment_stopped.set() + + def split_immediately(text: str) -> tuple[str, str]: + return text, "" + + pipeline = VoicePipeline( + workflow=FakeWorkflow([["first", "second"]]), + stt_model=FakeSTT(["user input"]), + tts_model=FailingTTS(), + config=VoicePipelineConfig(tts_settings=TTSModelSettings(text_splitter=split_immediately)), + ) + result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16))) + + with pytest.raises(RuntimeError, match="tts-failure"): + async for _event in result.stream(): + pass + + assert second_segment_stopped.is_set() + assert all(task.done() for task in result._tasks) + assert result._dispatcher_task is not None + assert result._dispatcher_task.done() + assert result._tracing_span is None + assert result.text_generation_task is not None + assert result.text_generation_task.done() + + @pytest.mark.asyncio async def test_streamed_audio_dispatcher_blocks_until_work_is_available() -> None: """The dispatcher must block while idle without losing a pre-wait notification."""