Python: Add opt-in AG-UI thread snapshot persistence and hydration

.github/.linkspector.yml

@@ -20,7 +20,10 @@ ignorePatterns:
   - pattern: "https://your-resource.openai.azure.com/"
   - pattern: "http://host.docker.internal"
   - pattern: "https://openai.github.io/openai-agents-js/openai/agents/classes/"
-  - pattern: "https:\/\/dotnet.microsoft.com\/download"
+  # dotnet.microsoft.com bot-blocks CI link checkers with intermittent 403s on any
+  # path (including localized variants like /en-us/download/...), so ignore the
+  # whole domain rather than just /download.
+  - pattern: "https:\/\/dotnet.microsoft.com"
   - pattern: "https://github.com/Rel1cx/eslint-react"
 # excludedDirs:
   # Folders which include links to localhost, since it's not ignored with regular expressions

python/packages/ag-ui/AGENTS.md

@@ -10,10 +10,12 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 - **`AGUIHttpService`** - HTTP service for AG-UI endpoints
 - **`AGUIEventConverter`** - Converts between Agent Framework and AG-UI events
 - **`add_agent_framework_fastapi_endpoint()`** - Add AG-UI endpoint to FastAPI app (`SupportsAgentRun` or `Workflow`)
+- **`InMemoryAGUIThreadSnapshotStore`** - Memory-only latest AG-UI Thread Snapshot store for local development, demos, and tests
 
 ## Types
 
 - **`AGUIRequest`** / **`AGUIChatOptions`** - Request types
+- **`AGUIThreadSnapshot`** / **`AGUIThreadSnapshotStore`** - Replayable thread snapshot model and scoped async store protocol
 - **`availableInterrupts` / `resume`** - Optional interrupt configuration and continuation payloads
 - **`AgentState`** / **`RunMetadata`** - State management types
 - **`PredictStateConfig`** - Configuration for state prediction

python/packages/ag-ui/README.md

@@ -198,6 +198,71 @@ The `dependencies` parameter accepts any FastAPI dependency, enabling integratio
 
 For a complete authentication example, see [getting_started/server.py](getting_started/server.py).
 
+## AG-UI Thread Snapshots
+
+AG-UI Thread Snapshot persistence is opt-in and disabled by default. Existing endpoints keep their current behavior
+unless you provide a `snapshot_store`.
+
+Thread snapshots let an AG-UI frontend recover replayable UI state after a refresh. When snapshot persistence is
+enabled, the endpoint stores the latest replayable snapshot for an AG-UI Thread within an application-defined
+Snapshot Scope. A Hydrate Request is an AG-UI request with a known `threadId`, `messages: []`, and no `resume`
+payload. Hydration replays the stored Shared State, message snapshot, and interruption metadata when available,
+then finishes without invoking the wrapped agent or workflow.
+
+Use the built-in in-memory store for local development, demos, and tests:
+
+```python
+from fastapi import FastAPI
+
+from agent_framework.ag_ui import InMemoryAGUIThreadSnapshotStore, add_agent_framework_fastapi_endpoint
+
+app = FastAPI()
+agent = ...
+snapshot_store = InMemoryAGUIThreadSnapshotStore(max_snapshots=500)
+
+
+def resolve_snapshot_scope(request):
+    # Local demo scope. Production apps should derive the scope from authenticated user or tenant context.
+    del request
+    return "local-demo"
+
+
+add_agent_framework_fastapi_endpoint(
+    app,
+    agent,
+    "/",
+    snapshot_store=snapshot_store,
+    snapshot_scope_resolver=resolve_snapshot_scope,
+)
+```
+
+A frontend can then hydrate the latest stored snapshot for the scoped thread:
+
+```json
+{
+  "threadId": "thread-1",
+  "messages": []
+}
+```
+
+Endpoint configuration requires `snapshot_scope_resolver` whenever a snapshot store is configured, including when
+the store is already set on a pre-wrapped `AgentFrameworkAgent` or `AgentFrameworkWorkflow`. The resolver returns
+the application-defined Snapshot Scope used with the AG-UI Thread id as the storage key.
+
+AG-UI Thread ids identify AG-UI Threads; they do not authorize snapshot access. Do not treat a thread id as a bearer
+credential or tenant boundary. Production applications must authenticate and authorize every AG-UI endpoint request
+and choose a Snapshot Scope that represents the app's real access boundary, such as an authenticated user, tenant,
+or workspace. Do not rely on untrusted client-provided fields by themselves to choose that boundary.
+
+Stored snapshots are untrusted application data with confidentiality impact. They may contain sensitive user text,
+model output, tool results, function arguments, UI payloads, Shared State, and interruption data. The built-in
+`InMemoryAGUIThreadSnapshotStore` is in-memory only, process-local, bounded, latest-only, and not durable production
+storage. It is cleared on process restart and is not shared across workers.
+
+No file-backed AG-UI snapshot store is provided by the package. Applications that need durable persistence should
+provide an app-owned implementation of the `AGUIThreadSnapshotStore` protocol and own storage hardening, including
+encryption, access control, retention, audit, data residency, and deletion behavior.
+
 ## Architecture
 
 The package uses a clean, orchestrator-based architecture:

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

@@ -9,6 +9,15 @@
 from ._endpoint import add_agent_framework_fastapi_endpoint
 from ._event_converters import AGUIEventConverter
 from ._http_service import AGUIHttpService
+from ._snapshots import (
+    DEFAULT_MAX_THREAD_SNAPSHOTS,
+    AGUIThreadID,
+    AGUIThreadSnapshot,
+    AGUIThreadSnapshotStore,
+    InMemoryAGUIThreadSnapshotStore,
+    SnapshotScope,
+    SnapshotScopeResolver,
+)
 from ._state import state_update
 from ._types import AgentState, AGUIChatOptions, AGUIRequest, PredictStateConfig, RunMetadata
 from ._workflow import AgentFrameworkWorkflow, WorkflowFactory
@@ -31,9 +40,16 @@
     "AGUIEventConverter",
     "AGUIHttpService",
     "AGUIRequest",
+    "AGUIThreadID",
+    "AGUIThreadSnapshot",
+    "AGUIThreadSnapshotStore",
     "AgentState",
+    "InMemoryAGUIThreadSnapshotStore",
     "PredictStateConfig",
     "RunMetadata",
+    "SnapshotScope",
+    "SnapshotScopeResolver",
+    "DEFAULT_MAX_THREAD_SNAPSHOTS",
     "DEFAULT_TAGS",
     "state_update",
     "__version__",

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

@@ -10,6 +10,7 @@
 from agent_framework import SupportsAgentRun
 
 from ._agent_run import PendingApprovalEntry, run_agent_stream
+from ._snapshots import AGUIThreadSnapshotStore
 
 
 class AgentConfig:
@@ -21,6 +22,7 @@ def __init__(
         predict_state_config: dict[str, dict[str, str]] | None = None,
         use_service_session: bool = False,
         require_confirmation: bool = True,
+        snapshot_store: AGUIThreadSnapshotStore | None = None,
     ):
         """Initialize agent configuration.
 
@@ -29,11 +31,14 @@ def __init__(
             predict_state_config: Configuration for predictive state updates
             use_service_session: Whether the agent session is service-managed
             require_confirmation: Whether predictive updates require user confirmation before applying
+            snapshot_store: Optional AG-UI Thread Snapshot store. Snapshot persistence remains inactive unless
+                endpoint setup also provides an explicit Snapshot Scope resolver.
         """
         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
 
     @staticmethod
     def _normalize_state_schema(state_schema: Any | None) -> dict[str, Any]:
@@ -79,6 +84,7 @@ def __init__(
         predict_state_config: dict[str, dict[str, str]] | None = None,
         require_confirmation: bool = True,
         use_service_session: bool = False,
+        snapshot_store: AGUIThreadSnapshotStore | None = None,
     ):
         """Initialize the AG-UI compatible agent wrapper.
 
@@ -90,6 +96,8 @@ def __init__(
             predict_state_config: Configuration for predictive state updates
             require_confirmation: Whether predictive updates require user confirmation before applying
             use_service_session: Whether the agent session is service-managed
+            snapshot_store: Optional AG-UI Thread Snapshot store. Snapshot persistence remains inactive unless
+                endpoint setup also provides an explicit Snapshot Scope resolver.
         """
         self.agent = agent
         self.name = name or getattr(agent, "name", "agent")
@@ -100,6 +108,7 @@ def __init__(
             predict_state_config=predict_state_config,
             use_service_session=use_service_session,
             require_confirmation=require_confirmation,
+            snapshot_store=snapshot_store,
         )
 
         # Server-side registry of pending approval requests.
@@ -110,6 +119,11 @@ def __init__(
         self._pending_approvals: OrderedDict[str, PendingApprovalEntry] = OrderedDict()
         self._pending_approvals_max_size: int = 10_000
 
+    @property
+    def snapshot_store(self) -> AGUIThreadSnapshotStore | None:
+        """Configured AG-UI Thread Snapshot store, if any."""
+        return self.config.snapshot_store
+
     async def run(
         self,
         input_data: dict[str, Any],