Python: Support Responses API server-side compaction with store=false

python/PACKAGE_STATUS.md

@@ -143,6 +143,14 @@ listed below.
 - `agent-framework-core`: `FunctionInvocationContext.add_tools` and
   `FunctionInvocationContext.remove_tools` from `agent_framework/_middleware.py`
 
+#### `SERVER_COMPACTION`
+
+- `agent-framework-core`: `Content.from_compaction` and the `compaction` content type from
+  `agent_framework/_types.py`
+- `agent-framework-openai`: `ContextManagementOptions` and the `context_management` option on
+  `OpenAIChatOptions` (also used by `FoundryChatClient`), including replay of compaction items for
+  `store=False` requests
+
 #### `SESSION_STORE`
 
 - `agent-framework-core`: `SessionStore` and `FileSessionStore` from

python/packages/core/AGENTS.md

@@ -74,6 +74,9 @@ agent_framework/
   constructors are experimental under `COMPUTER_USE`; the rest of `Content` retains its existing stage. Computer
   results can omit screenshots in core; OpenAI-based connectors require them when converting to Responses items.
   Completed call/result pairs remain in the transcript but are not user-input requests.
+- **Server-side compaction** - `Content.from_compaction` and the `compaction` content type are experimental under
+  `SERVER_COMPACTION`. The content carries a provider's opaque compacted conversation state in `protected_data`;
+  how it composes with `CompactionStrategy` / `CompactionProvider` is not settled yet.
 - **`ChatOptions`** - TypedDict for chat request options
 
 ### Tools (`_tools.py`)

python/packages/core/agent_framework/_feature_stage.py

@@ -63,6 +63,7 @@ class ExperimentalFeature(str, Enum):
     MCP_LONG_RUNNING_TASKS = "MCP_LONG_RUNNING_TASKS"
     MCP_SKILLS = "MCP_SKILLS"
     PROGRESSIVE_TOOLS = "PROGRESSIVE_TOOLS"
+    SERVER_COMPACTION = "SERVER_COMPACTION"
     SESSION_STORE = "SESSION_STORE"
     TO_PROMPT_AGENT = "TO_PROMPT_AGENT"
     VECTOR_STORES = "VECTOR_STORES"

python/packages/core/agent_framework/_types.py

@@ -387,6 +387,7 @@ def _restore_compaction_annotation_in_additional_properties(
     "function_approval_request",
     "function_approval_response",
     "oauth_consent_request",
+    "compaction",
 ]
 
 
@@ -1505,6 +1506,43 @@ def from_oauth_consent_request(
             raw_representation=raw_representation,
         )
 
+    @classmethod
+    @experimental(feature_id=ExperimentalFeature.SERVER_COMPACTION)
+    def from_compaction(
+        cls: type[ContentT],
+        *,
+        protected_data: str,
+        id: str | None = None,
+        annotations: Sequence[Annotation] | None = None,
+        additional_properties: MutableMapping[str, Any] | None = None,
+        raw_representation: Any = None,
+    ) -> ContentT:
+        """Create compaction content.
+
+        A compaction item is an opaque, provider-encrypted summary of earlier conversation
+        state, produced by service-side context compaction (for example the OpenAI Responses
+        API ``context_management`` option). It carries that state into later turns in fewer
+        tokens, and providers that support it replay it as input on the next request.
+
+        Keyword Args:
+            protected_data: The provider's encrypted compaction payload.
+            id: The provider's identifier for the compaction item.
+            annotations: Optional annotations.
+            additional_properties: Optional additional properties.
+            raw_representation: Optional raw representation from the provider.
+
+        Returns:
+            A new Content instance with type ``compaction``.
+        """
+        return cls(
+            "compaction",
+            id=id,
+            protected_data=protected_data,
+            annotations=annotations,
+            additional_properties=additional_properties,
+            raw_representation=raw_representation,
+        )
+
     def to_function_approval_response(
         self,
         approved: bool,

python/packages/core/tests/core/test_feature_stage.py

@@ -82,6 +82,20 @@ def test_computer_use_content_is_experimental_without_staging_all_content() -> N
     assert "[COMPUTER_USE]" in str(caught[0].message)
 
 
+def test_server_compaction_content_is_experimental() -> None:
+    constructor = Content.from_compaction
+    assert getattr(constructor, "__feature_stage__", None) == "experimental"
+    assert getattr(constructor, "__feature_id__", None) == PublicExperimentalFeature.SERVER_COMPACTION.value
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always", ExperimentalWarning)
+        content = Content.from_compaction(id="cmp_1", protected_data="state")
+
+    assert content.type == "compaction"
+    assert len(caught) == 1
+    assert "[SERVER_COMPACTION]" in str(caught[0].message)
+
+
 def test_experimental_decorator_accepts_feature_enum() -> None:
     with warnings.catch_warnings(record=True) as caught:
         warnings.simplefilter("always")

python/packages/core/tests/core/test_types.py

@@ -5982,6 +5982,18 @@ def test_oauth_consent_request_serialization_roundtrip():
     assert d["user_input_request"] is True
 
 
+def test_compaction_content_creation_and_serialization_roundtrip():
+    """Test Content.from_compaction keeps the encrypted payload and id through serialization."""
+    content = Content.from_compaction(id="cmp_1", protected_data="encrypted-state")
+
+    assert content.type == "compaction"
+    assert content.id == "cmp_1"
+    assert content.protected_data == "encrypted-state"
+
+    restored = Content.from_dict(content.to_dict())
+    assert (restored.type, restored.id, restored.protected_data) == ("compaction", "cmp_1", "encrypted-state")
+
+
 # endregion
 
 

python/packages/foundry/tests/foundry/test_foundry_chat_client.py

@@ -2224,3 +2224,27 @@ async def test_shared_chat_client_keeps_concurrent_agent_runs_isolated(
     assert first_session.service_session_id == "response-first"
     assert second_session.service_session_id == "response-second"
     assert transport.max_active_requests == 2
+
+
+async def test_foundry_chat_client_replays_compaction_item_when_store_false() -> None:
+    """FoundryChatClient forwards context_management and replays the latest compaction item statelessly."""
+    mock_project_client = MagicMock()
+    mock_project_client.get_openai_client.return_value = _make_mock_openai_client()
+    client = FoundryChatClient(project_client=mock_project_client, model=_TEST_FOUNDRY_MODEL)
+    context_management = [{"type": "compaction", "compact_threshold": 200000}]
+
+    _, run_options, _ = await client._prepare_request(
+        [
+            Message(role="user", contents=["Earlier question"]),
+            Message(
+                role="assistant",
+                contents=[Content.from_compaction(id="cmp_1", protected_data="state"), "Earlier answer"],
+            ),
+            Message(role="user", contents=["Next question"]),
+        ],
+        {"store": False, "context_management": context_management},
+    )
+
+    assert run_options["context_management"] == context_management
+    assert run_options["input"][0] == {"type": "compaction", "encrypted_content": "state", "id": "cmp_1"}
+    assert "Earlier question" not in str(run_options["input"])

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -43,7 +43,10 @@
     GROUP_ID_KEY,
     CompactionStrategy,
     TokenizerProtocol,
+    included_messages,
+    set_excluded,
 )
+from agent_framework._feature_stage import ExperimentalFeature, experimental
 from agent_framework._middleware import ChatAndFunctionMiddlewareTypes, ChatMiddlewareLayer
 from agent_framework._settings import SecretString
 from agent_framework._telemetry import USER_AGENT_KEY, mark_feature_used
@@ -204,6 +207,17 @@ class ReasoningOptions(TypedDict, total=False):
     """How to summarize reasoning in the response."""
 
 
+@experimental(feature_id=ExperimentalFeature.SERVER_COMPACTION)
+class ContextManagementOptions(TypedDict, total=False):
+    """Service-side context management entry for the Responses API."""
+
+    type: Literal["compaction"]
+    """The context management strategy. Only ``compaction`` is supported."""
+
+    compact_threshold: int
+    """Token count at which the service compacts the context."""
+
+
 class StreamOptions(TypedDict, total=False):
     """Options for streaming responses."""
 
@@ -258,6 +272,14 @@ class OpenAIChatOptions(ChatOptions[ResponseFormatT], Generic[ResponseFormatT],
     Sending this option requires openai 2.45.0 or later.
     See: https://developers.openai.com/api/docs/guides/prompt-caching#prompt-cache-breakpoints"""
 
+    context_management: list[ContextManagementOptions]
+    """Service-side context management, such as automatic compaction.
+    When the context crosses ``compact_threshold`` the service emits an encrypted
+    compaction item, surfaced as ``compaction`` content. With ``store=False`` the
+    latest compaction item is replayed on later requests and the history before it
+    (other than system/developer messages) is no longer sent.
+    See: https://learn.microsoft.com/azure/foundry/openai/how-to/responses#server-side-compaction"""
+
     reasoning: ReasoningOptions
     """Configuration for reasoning models (gpt-5, o-series).
     See: https://platform.openai.com/docs/guides/reasoning"""
@@ -1749,6 +1771,7 @@ def _prepare_messages_for_openai(
         """
         reasoning_items: dict[str, dict[str, Any]] = {}
         if not request_uses_service_side_storage:
+            chat_messages = self._exclude_messages_before_latest_compaction(chat_messages)
             reasoning_items = self._prepare_reasoning_items_for_openai(chat_messages)
             self._validate_reasoning_groups_for_stateless_replay(
                 chat_messages,
@@ -1771,6 +1794,34 @@ def _prepare_messages_for_openai(
         # items (drop unmatched). See `_AF_MCP_PENDING_OUTPUT_KEY`.
         return self._coalesce_pending_mcp_results(flat)
 
+    @staticmethod
+    def _exclude_messages_before_latest_compaction(chat_messages: Sequence[Message]) -> Sequence[Message]:
+        """Mark the history that a service-side compaction item already summarizes as excluded.
+
+        The latest compaction item with a replayable payload carries the context needed to
+        continue, so for stateless replay the messages before the one that contains it are marked
+        excluded rather than removed, matching how the built-in compaction strategies work. System
+        and developer messages are kept because they carry instructions rather than conversation
+        state. Whole messages are marked or kept so tool calls stay paired with their results.
+        Returns the messages that remain included.
+        """
+        latest_index = next(
+            (
+                index
+                for index in range(len(chat_messages) - 1, -1, -1)
+                if any(
+                    content.type == "compaction" and content.protected_data for content in chat_messages[index].contents
+                )
+            ),
+            None,
+        )
+        if latest_index is None or latest_index == 0:
+            return chat_messages
+        for message in chat_messages[:latest_index]:
+            if message.role not in ("system", "developer"):
+                set_excluded(message, excluded=True, reason="superseded by service-side compaction")
+        return included_messages(list(chat_messages))
+
     def _validate_reasoning_groups_for_stateless_replay(
         self,
         chat_messages: Sequence[Message],
@@ -1905,6 +1956,20 @@ def _prepare_message_for_openai(
                         all_messages.append(reasoning_item)
                         serialized_reasoning_ids.add(content.id)
                     continue
+                case "compaction":
+                    # With service-side storage the service already holds the compaction item.
+                    if request_uses_service_side_storage or not content.protected_data:
+                        continue
+                    if "content" in args or "tool_calls" in args:
+                        all_messages.append(args)
+                        args = {"type": "message", "role": message.role}
+                    compaction_item: dict[str, Any] = {
+                        "type": "compaction",
+                        "encrypted_content": content.protected_data,
+                    }
+                    if content.id:
+                        compaction_item["id"] = content.id
+                    all_messages.append(compaction_item)
                 case "function_result":
                     new_args: dict[str, Any] = {}
                     new_args.update(
@@ -3264,6 +3329,23 @@ def _get_finish_reason_from_openai_response(self, response: Any) -> FinishReason
             return FinishReason("tool_calls")
         return FinishReason("stop")
 
+    @staticmethod
+    def _parse_compaction_content(item: Any) -> Content | None:
+        """Convert a Responses API compaction item into compaction content."""
+        encrypted_content = getattr(item, "encrypted_content", None)
+        if not isinstance(encrypted_content, str) or not encrypted_content:
+            logger.debug("Ignoring compaction item without encrypted content: %s", item)
+            return None
+        additional_properties: dict[str, Any] = {}
+        if created_by := getattr(item, "created_by", None):
+            additional_properties["created_by"] = created_by
+        return Content.from_compaction(
+            id=getattr(item, "id", None),
+            protected_data=encrypted_content,
+            additional_properties=additional_properties or None,
+            raw_representation=item,
+        )
+
     def _parse_response_from_openai(
         self,
         response: OpenAIResponse | ParsedResponse[BaseModel],
@@ -3539,6 +3621,9 @@ def _parse_response_from_openai(
                     contents.extend(self._image_generation_item_to_contents(item))
                 case "shell_call" | "local_shell_call" | "shell_call_output":
                     contents.extend(self._shell_item_to_contents(item, local_shell_tool_name))
+                case "compaction":  # ResponseCompactionItem
+                    if compaction_content := self._parse_compaction_content(item):
+                        contents.append(compaction_content)
                 case _:
                     logger.debug("Unparsed output of type: %s: %s", item.type, item)
         self._mark_completed_computer_calls(contents)
@@ -4123,6 +4208,9 @@ def parse_annotations(
                                 raw_representation=done_item,
                             )
                         )
+                elif getattr(done_item, "type", None) == "compaction":
+                    if compaction_content := self._parse_compaction_content(done_item):
+                        contents.append(compaction_content)
                 elif done_item.type == "message":
                     for content_index, part in enumerate(done_item.content):
                         if part.type == "output_text":