Add test for correct streaming chunks and responses id consistency

tests/mcp_tests/test_aresponses_api_with_mcp.py

@@ -1250,4 +1250,147 @@ async def test_no_duplicate_mcp_tools_in_streaming_e2e():
         }
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model", ["gpt-4o-mini"])
+async def test_streaming_mcp_event_order_and_response_id_consistency(
+    model: str, caplog: pytest.LogCaptureFixture
+):
+    """
+    Test that:
+    1. Streaming events are emitted in correct order (response.created, response.in_progress, response.output_item.added before MCP events)
+    2. All response lifecycle events share the same response ID within a cycle
+    """
+    if ("gpt" in model.lower() or "openai" in model.lower()) and not os.getenv("OPENAI_API_KEY"):
+        pytest.skip("OPENAI_API_KEY not set, skipping openai model test")
+
+    from unittest.mock import AsyncMock, patch
+
+    mock_mcp_tools = [
+        type('MCPTool', (), {
+            'name': 'get_weather',
+            'description': 'Get weather for a city',
+            'inputSchema': {
+                "type": "object", 
+                "properties": {
+                    "city": {"type": "string", "description": "City name"}
+                },
+                "required": ["city"]
+            }
+        })()
+    ]
+
+    with caplog.at_level(logging.ERROR):
+        with patch.object(
+            LiteLLM_Proxy_MCP_Handler,
+            '_get_mcp_tools_from_manager',
+            new_callable=AsyncMock,
+        ) as mock_get_tools, patch.object(
+            LiteLLM_Proxy_MCP_Handler,
+            '_execute_tool_calls',
+            new_callable=AsyncMock,
+        ) as mock_execute_tools:
+            mock_get_tools.return_value = (mock_mcp_tools, ["litellm_proxy"])
+
+            def mock_execute_side_effect(tool_calls, user_api_key_auth, **kwargs):
+                results = []
+                for tool_call in tool_calls:
+                    call_id = None
+                    if isinstance(tool_call, dict):
+                        call_id = tool_call.get("call_id") or tool_call.get("id")
+                    elif hasattr(tool_call, 'call_id'):
+                        call_id = tool_call.call_id
+                    elif hasattr(tool_call, 'id'):
+                        call_id = tool_call.id
+                    if call_id:
+                        results.append({
+                            "tool_call_id": call_id,
+                            "result": "Sunny, 72°F",
+                        })
+                return results
+
+            mock_execute_tools.side_effect = mock_execute_side_effect
+
+            mcp_tool_config = cast(Any, {
+                "type": "mcp",
+                "server_url": "litellm_proxy",
+                "require_approval": "never",
+            })
+
+            response = await litellm.aresponses(
+                model=model,
+                tools=[mcp_tool_config],
+                input=[{
+                    "role": "user",
+                    "type": "message",
+                    "content": "What's the weather in San Francisco?"
+                }],
+                stream=True,
+            )
+
+            events = []
+            async for chunk in response:
+                events.append(chunk)
+
+            assert len(events) > 0, "Should receive streaming events"
+
+            created_idx = next((i for i, e in enumerate(events) if getattr(e, 'type', None) == 'response.created'), None)
+            in_progress_idx = next((i for i, e in enumerate(events) if getattr(e, 'type', None) == 'response.in_progress'), None)
+            output_item_added_idx = next((i for i, e in enumerate(events) if getattr(e, 'type', None) == 'response.output_item.added'), None)
+            mcp_in_progress_idx = next((i for i, e in enumerate(events) if 'mcp_list_tools.in_progress' in str(getattr(e, 'type', ''))), None)
+            completed_idx = next((i for i, e in enumerate(events) if getattr(e, 'type', None) == 'response.completed'), None)
+
+            assert created_idx is not None, "response.created event should be present"
+            assert in_progress_idx is not None, "response.in_progress event should be present"
+            assert output_item_added_idx is not None, "response.output_item.added event should be present"
+
+            assert created_idx < in_progress_idx, "response.created should come before response.in_progress"
+            assert in_progress_idx < output_item_added_idx, "response.in_progress should come before response.output_item.added"
+
+            if mcp_in_progress_idx is not None:
+                assert output_item_added_idx < mcp_in_progress_idx, "response.output_item.added should come before response.mcp_list_tools.in_progress"
+
+            response_ids = []
+            for i, event in enumerate(events):
+                event_type = getattr(event, 'type', None)
+                if hasattr(event, 'response'):
+                    response_obj = getattr(event, 'response', None)
+                    if response_obj and hasattr(response_obj, 'id'):
+                        event_type_value = event_type.value if hasattr(event_type, 'value') else str(event_type)
+                        if any(x in event_type_value for x in ['response.created', 'response.in_progress', 'response.completed']):
+                            response_ids.append((i, event_type_value, response_obj.id))
+
+            assert len(response_ids) >= 2, f"Should have at least 2 response lifecycle events. Found {len(response_ids)}"
+
+            cycles = []
+            current_cycle = []
+            current_id = None
+
+            for idx, event_type, resp_id in response_ids:
+                if current_id is None or resp_id == current_id:
+                    current_cycle.append((idx, event_type, resp_id))
+                    current_id = resp_id
+                else:
+                    if current_cycle:
+                        cycles.append(current_cycle)
+                    current_cycle = [(idx, event_type, resp_id)]
+                    current_id = resp_id
+            if current_cycle:
+                cycles.append(current_cycle)
+
+            for cycle_num, cycle in enumerate(cycles):
+                cycle_ids = set(resp_id for _, _, resp_id in cycle)
+                assert len(cycle_ids) == 1, f"Cycle {cycle_num + 1} should have consistent response ID. Found {len(cycle_ids)} unique IDs"
+
+            assert completed_idx is not None, "response.completed event should be present"
+
+    lite_errors = [
+        record for record in caplog.records
+        if record.levelno >= logging.ERROR
+        and ("LiteLLM" in record.name or "LiteLLM" in record.getMessage())
+    ]
+    assert not lite_errors, "Unexpected LiteLLM errors: " + ", ".join(
+        record.getMessage() for record in lite_errors
+    )
+
+