Python: feat(ag-ui): add emit_messages_snapshot config to suppress terminal snapshot

python/packages/ag-ui/agent_framework_ag_ui/_agent.py

@@ -26,6 +26,7 @@ def __init__(
         require_confirmation: bool = True,
         snapshot_store: AGUIThreadSnapshotStore | None = None,
         a2ui_config: dict[str, Any] | None = None,
+        emit_messages_snapshot: bool = True,
     ):
         """Initialize agent configuration.
 
@@ -42,13 +43,17 @@ def __init__(
             require_confirmation: Whether predictive updates require user confirmation before applying
             a2ui_config: Optional backend A2UI config consumed by auto-injection
                 (``forwardedProps.injectA2UITool``). See ``plan_a2ui_injection``.
+            emit_messages_snapshot: Whether to emit a terminal MessagesSnapshotEvent at the end of runs.
+                Defaults to True for backward compatibility. Set to False when using HistoryProvider
+                to prevent redundant full-transcript rewrites on the client.
         """
         self.state_schema = self._normalize_state_schema(state_schema)
         self.predict_state_config = predict_state_config or {}
         self.use_service_session = use_service_session
         self.require_confirmation = require_confirmation
         self.snapshot_store = snapshot_store
         self.a2ui_config = a2ui_config
+        self.emit_messages_snapshot = emit_messages_snapshot
 
     @staticmethod
     def _normalize_state_schema(state_schema: Any | None) -> dict[str, Any]:
@@ -96,6 +101,7 @@ def __init__(
         use_service_session: bool = False,
         snapshot_store: AGUIThreadSnapshotStore | None = None,
         a2ui_config: dict[str, Any] | None = None,
+        emit_messages_snapshot: bool = True,
     ):
         """Initialize the AG-UI compatible agent wrapper.
 
@@ -113,6 +119,8 @@ def __init__(
             snapshot_store: Optional AG-UI Thread Snapshot store. Snapshot persistence remains inactive unless
                 endpoint setup also provides an explicit Snapshot Scope resolver.
             a2ui_config: Optional backend A2UI config consumed by auto-injection.
+            emit_messages_snapshot: Whether to emit a terminal MessagesSnapshotEvent at the end of runs.
+                Defaults to True. Set to False when using HistoryProvider.
         """
         self.agent = agent
         self.name = name or getattr(agent, "name", "agent")
@@ -125,6 +133,7 @@ def __init__(
             require_confirmation=require_confirmation,
             snapshot_store=snapshot_store,
             a2ui_config=a2ui_config,
+            emit_messages_snapshot=emit_messages_snapshot,
         )
 
         # Server-side Approval State. Populated when approval requests are emitted

python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py

@@ -2952,7 +2952,7 @@ async def run_agent_stream(
         # so an unrelated user tool named "generate_a2ui" keeps its snapshot.
         if a2ui_active:
             logger.info("Suppressing terminal MessagesSnapshotEvent for A2UI run to preserve streamed message order.")
-        if not a2ui_active and not _should_suppress_intermediate_snapshot(
+        if config.emit_messages_snapshot and not a2ui_active and not _should_suppress_intermediate_snapshot(
             last_tool_name, predict_state_config, config.require_confirmation
         ):
             yield snapshot_event

python/packages/ag-ui/tests/ag_ui/test_run.py

@@ -1823,6 +1823,44 @@ async def test_run_agent_stream_accumulates_multiple_confirm_interrupts():
     assert interrupt_tool_names == {"generate_tasks", "generate_notes"}
 
 
+async def test_run_agent_stream_suppresses_messages_snapshot_if_configured():
+    """When emit_messages_snapshot=False, no terminal MessagesSnapshotEvent is yielded."""
+    from conftest import StubAgent  # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports]
+
+    from agent_framework_ag_ui import AgentFrameworkAgent
+
+    updates = [
+        AgentResponseUpdate(contents=[Content.from_text("Hello")], role="assistant"),
+    ]
+
+    stub = StubAgent(updates=updates)
+    agent = AgentFrameworkAgent(
+        agent=stub,
+        emit_messages_snapshot=False,
+    )
+
+    payload = {
+        "thread_id": "thread-1",
+        "run_id": "run-1",
+        "messages": [{"role": "user", "content": "Hi"}],
+    }
+
+    events = [event async for event in agent.run(payload)]
+
+    # We should have TextMessageStart/Delta/End, but no MessagesSnapshot
+    snapshot_events = [e for e in events if getattr(e, "type", None) == "MESSAGES_SNAPSHOT"]
+    assert len(snapshot_events) == 0, "MessagesSnapshotEvent should be suppressed"
+
+    # Still finishes normally
+    finished_events = [
+        e
+        for e in events
+        if getattr(e, "type", None) == "RUN_FINISHED"
+        or getattr(getattr(e, "type", None), "value", None) == "RUN_FINISHED"
+    ]
+    assert len(finished_events) == 1
+
+
 def test_emit_oauth_consent_request():
     """Test that oauth_consent_request content emits a CustomEvent."""
     content = Content.from_oauth_consent_request(