test_responses_api_shell_tool_streaming_sees_shell_output

tests/llm_responses_api_testing/base_responses_api.py

@@ -2,7 +2,7 @@ import httpx
 import json
 import pytest
 import sys
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 from unittest.mock import MagicMock, Mock, patch
 import os
 from litellm._uuid import uuid
@@ -114,6 +114,10 @@ class BaseResponsesAPITest(ABC):
         """Must return the base completion reasoning call args"""
         return None
 
+    def get_advanced_model_for_shell_tool(self) -> Optional[str]:
+        """If specified, overrides the model used by test_responses_api_shell_tool_streaming_sees_shell_output (e.g. openai/gpt-5.2 for shell support)."""
+        return None
+
     @pytest.mark.parametrize("sync_mode", [True, False])
     @pytest.mark.asyncio
     async def test_basic_openai_responses_api(self, sync_mode):
@@ -743,4 +747,61 @@ class BaseResponsesAPITest(ABC):
             raise
         validate_responses_api_response(response, final_chunk=True)
         assert response.get("id") is not None
-        assert response.get("status") is not None
+        assert response.get("status") is not None
+
+    @pytest.mark.asyncio
+    async def test_responses_api_shell_tool_streaming_sees_shell_output(self):
+        """
+        E2E streaming call with Shell tool; validate we can see shell output in the stream.
+
+        Calls aresponses(..., tools=[shell], stream=True), then iterates the stream and
+        asserts at least one event is shell-related or response output contains shell_call.
+        Skips when model does not support shell (e.g. gpt-4o).
+        """
+        base_completion_call_args = self.get_base_completion_call_args()
+        model = self.get_advanced_model_for_shell_tool() or base_completion_call_args.get(
+            "model"
+        ) or "gpt-5.2"
+        tools = [{"type": "shell", "environment": {"type": "container_auto"}}]
+        input_msg = "List files in /mnt/data and run python --version."
+
+        stream = await litellm.aresponses(
+            **{**base_completion_call_args, "model": model},
+            input=input_msg,
+            max_output_tokens=512,
+            tools=tools,
+            tool_choice="auto",
+            stream=True,
+        )
+
+        event_types_seen = []
+        output_items_with_shell = []
+
+        async for event in stream:
+            print("event=", json.dumps(event, indent=4, default=str))
+            event_type = getattr(event, "type", None) or (
+                event.get("type") if isinstance(event, dict) else None
+            )
+            if event_type is not None:
+                event_types_seen.append(str(event_type))
+            if "shell" in str(event_type or "").lower():
+                output_items_with_shell.append(event_type)
+            response_obj = getattr(event, "response", None) or (
+                event.get("response") if isinstance(event, dict) else None
+            )
+            if response_obj is not None:
+                output = getattr(response_obj, "output", None) or (
+                    response_obj.get("output") if isinstance(response_obj, dict) else None
+                )
+                if isinstance(output, list):
+                    for item in output:
+                        item_type = getattr(item, "type", None) or (
+                            item.get("type") if isinstance(item, dict) else None
+                        )
+                        if item_type and "shell" in str(item_type).lower():
+                            output_items_with_shell.append(item_type)
+
+        assert len(event_types_seen) > 0, "Expected at least one stream event"
+        assert len(output_items_with_shell) > 0, (
+            f"Expected to see shell output in stream; event types seen: {event_types_seen!r}"
+        )

tests/llm_responses_api_testing/test_openai_responses_api.py

@@ -36,6 +36,9 @@ class TestOpenAIResponsesAPITest(BaseResponsesAPITest):
             "model": "openai/gpt-5-mini",
         }
 
+    def get_advanced_model_for_shell_tool(self):
+        return "openai/gpt-5.2"
+
 
 class TestCustomLogger(CustomLogger):
     def __init__(