[Feat] Adds support for server-side compaction on the OpenAI Responses API context_management (#21058)

* test_responses_api_context_management_server_side_compaction

* Server-side compaction

* docs fix

* test_responses_api_shell_tool

docs/my-website/docs/response_api.md

@@ -1023,6 +1023,76 @@ curl http://localhost:4000/v1/responses \
 
 
 
+## Server-side compaction
+
+For long-running conversations, you can enable **server-side compaction** so that when the rendered context size crosses a threshold, the server automatically runs compaction in-stream and emits a compaction item—no separate `POST /v1/responses/compact` call is required.
+
+Supported on the OpenAI Responses API when using the `openai` or `azure` provider. Pass `context_management` with a compaction entry and `compact_threshold` (token count; minimum 1000). When the context crosses the threshold, the server compacts in-stream and continues. Chain turns with `previous_response_id` or by appending output items to your next input array. See [OpenAI Compaction guide](https://developers.openai.com/api/docs/guides/compaction) for details.
+
+For explicit control over when compaction runs, use the standalone compact endpoint (`POST /v1/responses/compact`) instead.
+
+### Python SDK
+
+```python showLineNumbers title="Server-side compaction with LiteLLM Python SDK"
+import litellm
+
+# Non-streaming: enable compaction when context exceeds 200k tokens
+response = litellm.responses(
+    model="openai/gpt-4o",
+    input="Your conversation input...",
+    context_management=[{"type": "compaction", "compact_threshold": 200000}],
+    max_output_tokens=1024,
+)
+print(response)
+
+# Streaming: same context_management, compaction runs in-stream if threshold is crossed
+stream = litellm.responses(
+    model="openai/gpt-4o",
+    input="Your conversation input...",
+    context_management=[{"type": "compaction", "compact_threshold": 200000}],
+    stream=True,
+)
+for event in stream:
+    print(event)
+```
+
+### LiteLLM Proxy (AI Gateway)
+
+Use the OpenAI SDK with your proxy as `base_url`, or call the proxy with curl. The proxy forwards `context_management` to the provider.
+
+**OpenAI Python SDK (proxy as base_url):**
+
+```python showLineNumbers title="Server-side compaction via LiteLLM Proxy"
+from openai import OpenAI
+
+client = OpenAI(
+    base_url="http://localhost:4000",  # LiteLLM Proxy (AI Gateway)
+    api_key="your-proxy-api-key",
+)
+
+response = client.responses.create(
+    model="openai/gpt-4o",
+    input="Your conversation input...",
+    context_management=[{"type": "compaction", "compact_threshold": 200000}],
+    max_output_tokens=1024,
+)
+print(response)
+```
+
+**curl (proxy):**
+
+```bash title="Server-side compaction via curl to LiteLLM Proxy"
+curl -X POST "http://localhost:4000/v1/responses" \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer your-proxy-api-key" \
+  -d '{
+    "model": "openai/gpt-4o",
+    "input": "Your conversation input...",
+    "context_management": [{"type": "compaction", "compact_threshold": 200000}],
+    "max_output_tokens": 1024
+  }'
+```
+
 ## Session Management
 
 LiteLLM Proxy supports session management for all supported models. This allows you to store and fetch conversation history (state) in LiteLLM Proxy. 

litellm/types/llms/openai.py

@@ -1074,6 +1074,19 @@ class PromptObject(TypedDict, total=False):
     """Optional version of the prompt template."""
 
 
+class ContextManagementEntry(TypedDict, total=False):
+    """
+    Context management configuration entry for a request.
+    See https://developers.openai.com/api/docs/guides/compaction.
+    """
+
+    type: str
+    """The context management entry type. Currently only ``'compaction'`` is supported."""
+
+    compact_threshold: int
+    """Token threshold at which compaction is triggered for this entry. Minimum 1000."""
+
+
 class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
     """TypedDict for Optional parameters supported by the responses API."""
 
@@ -1104,6 +1117,8 @@ class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
     partial_images: Optional[
         int
     ]  # Number of partial images to generate (1-3) for streaming image generation
+    context_management: Optional[List[ContextManagementEntry]]
+    """Context management configuration. E.g. [{\"type\": \"compaction\", \"compact_threshold\": 200000}] for server-side compaction (minimum 1000)."""
 
 
 class ResponsesAPIRequestParams(ResponsesAPIOptionalRequestParams, total=False):

tests/llm_responses_api_testing/base_responses_api.py

@@ -669,7 +669,7 @@ class BaseResponsesAPITest(ABC):
     async def test_cancel_responses_invalid_response_id(self, sync_mode):
         """Test cancel_responses with invalid response ID should raise appropriate error"""
         base_completion_call_args = self.get_base_completion_call_args()
-        
+
         if sync_mode:
             with pytest.raises(Exception):
                 litellm.cancel_responses(
@@ -679,4 +679,41 @@ class BaseResponsesAPITest(ABC):
             with pytest.raises(Exception):
                 await litellm.acancel_responses(
                     response_id="invalid_response_id_12345", **base_completion_call_args
-                )
+                )
+
+    @pytest.mark.parametrize("sync_mode", [True, False])
+    @pytest.mark.asyncio
+    async def test_responses_api_context_management_server_side_compaction(self, sync_mode):
+        """
+        E2E test for server-side compaction (context_management) on OpenAI Responses API.
+        Passes context_management with compact_threshold; validates that the request is
+        accepted and returns a valid response. Compaction may not run for short inputs.
+        """
+        base_completion_call_args = self.get_base_completion_call_args()
+        model = base_completion_call_args.get("model") or ""
+        # Only run with context_management for OpenAI (OAI) for now
+        if "openai/" not in str(model) and "azure/" not in str(model):
+            pytest.skip(
+                "context_management server-side compaction e2e is only run for OpenAI/Azure"
+            )
+        context_management = [{"type": "compaction", "compact_threshold": 200000}]
+        try:
+            if sync_mode:
+                response = litellm.responses(
+                    input="Short ping to verify context_management is accepted.",
+                    max_output_tokens=20,
+                    context_management=context_management,
+                    **base_completion_call_args,
+                )
+            else:
+                response = await litellm.aresponses(
+                    input="Short ping to verify context_management is accepted.",
+                    max_output_tokens=20,
+                    context_management=context_management,
+                    **base_completion_call_args,
+                )
+        except litellm.InternalServerError:
+            pytest.skip("Skipping test due to litellm.InternalServerError")
+        validate_responses_api_response(response, final_chunk=True)
+        assert response.get("id") is not None
+        assert response.get("status") is not None

tests/test_litellm/expected_responses_api_request/context_management_and_shell.json

@@ -0,0 +1,20 @@
+{
+  "model": "gpt-4o",
+  "input": "List files in /mnt/data and run python --version.",
+  "context_management": [
+    {
+      "type": "compaction",
+      "compact_threshold": 200000
+    }
+  ],
+  "tools": [
+    {
+      "type": "shell",
+      "environment": {
+        "type": "container_auto"
+      }
+    }
+  ],
+  "tool_choice": "auto",
+  "max_output_tokens": 256
+}

tests/test_litellm/responses/test_responses_api_request_body.py

@@ -0,0 +1,103 @@
+"""
+Test that litellm.responses() / litellm.aresponses() send the expected request body
+over the wire. Expected JSON bodies are stored in expected_responses_api_request/.
+"""
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import pytest
+
+import litellm
+
+
+def _expected_dir() -> Path:
+    """Path to expected_responses_api_request folder (sibling of test_litellm/responses)."""
+    return Path(__file__).resolve().parent.parent / "expected_responses_api_request"
+
+
+@pytest.mark.asyncio
+async def test_aresponses_context_management_and_shell_request_body_matches_expected():
+    """
+    Call litellm.aresponses() with context_management and shell tool;
+    assert the httpx POST request body matches the expected JSON.
+    """
+    expected_path = _expected_dir() / "context_management_and_shell.json"
+    assert expected_path.exists(), f"Expected file not found: {expected_path}"
+    with open(expected_path) as f:
+        expected_body = json.load(f)
+
+    # Minimal Responses API response so parsing succeeds
+    mock_response = {
+        "id": "resp_ctx_shell_test",
+        "object": "response",
+        "created_at": 1734366691,
+        "status": "completed",
+        "model": "gpt-4o",
+        "output": [
+            {
+                "type": "message",
+                "id": "msg_1",
+                "status": "completed",
+                "role": "assistant",
+                "content": [
+                    {"type": "output_text", "text": "Done.", "annotations": []}
+                ],
+            }
+        ],
+        "parallel_tool_calls": True,
+        "usage": {
+            "input_tokens": 10,
+            "output_tokens": 5,
+            "total_tokens": 15,
+            "output_tokens_details": {"reasoning_tokens": 0},
+        },
+        "error": None,
+        "incomplete_details": None,
+        "instructions": None,
+        "metadata": None,
+        "temperature": None,
+        "tool_choice": "auto",
+        "tools": [],
+        "top_p": None,
+        "max_output_tokens": None,
+        "previous_response_id": None,
+        "reasoning": None,
+        "truncation": None,
+        "user": None,
+    }
+
+    class MockResponse:
+        def __init__(self, json_data, status_code=200):
+            self._json_data = json_data
+            self.status_code = status_code
+            self.text = json.dumps(json_data)
+            self.headers = httpx.Headers({})
+
+        def json(self):
+            return self._json_data
+
+    with patch(
+        "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
+        new_callable=AsyncMock,
+    ) as mock_post:
+        mock_post.return_value = MockResponse(mock_response, 200)
+
+        await litellm.aresponses(
+            model="openai/gpt-4o",
+            input=expected_body["input"],
+            context_management=expected_body["context_management"],
+            tools=expected_body["tools"],
+            tool_choice=expected_body["tool_choice"],
+            max_output_tokens=expected_body["max_output_tokens"],
+        )
+
+        mock_post.assert_called_once()
+        request_body = mock_post.call_args.kwargs["json"]
+
+        for key, expected_value in expected_body.items():
+            assert key in request_body, f"Missing key in request body: {key}"
+            assert request_body[key] == expected_value, (
+                f"Mismatch for key {key}: got {request_body[key]!r}, expected {expected_value!r}"
+            )