Python: Foundry Hosted Agent Resiliency Support

python/packages/foundry_hosting/README.md

@@ -46,4 +46,4 @@ locally. Stored checkpoints are scoped under `checkpoints`.
 
 `ResponsesHostServer` persists function approvals durably. By default, it uses the
 `FoundryFunctionApprovalStore`, backed by Foundry storage when hosted and file-based
-storage locally. Stored approvals are scoped under `function_approvals`.
+storage locally. Stored approvals are scoped under `function_approvals`.

python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py

@@ -90,7 +90,6 @@ async def _get_store(self) -> FoundryStateStore:
         return await FoundryStateStore.get_or_create(
             f"{self.DEFAULT_ROOT_SCOPE}/{self.context_id}",
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID:
@@ -232,7 +231,6 @@ async def _get_store(self) -> FoundryStateStore:
         return await FoundryStateStore.get_or_create(
             self.DEFAULT_ROOT_SCOPE,
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def save_approval_request(self, approval_request_id: str, request: Content) -> None:
@@ -280,7 +278,6 @@ async def _get_store(self) -> FoundryStateStore:
         return await FoundryStateStore.get_or_create(
             f"{self.DEFAULT_ROOT_SCOPE}",
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def get(self, session_id: str) -> AgentSession | None:

python/packages/foundry_hosting/tests/test_responses.py

@@ -15,6 +15,7 @@
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Literal, cast, overload
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -47,8 +48,14 @@
     tool,
 )
 from azure.ai.agentserver.core import get_request_context
-from azure.ai.agentserver.responses import InMemoryResponseProvider, ResponseContext
+from azure.ai.agentserver.responses import (
+    FileResponseStore,
+    InMemoryResponseProvider,
+    ResponseContext,
+    ResponsesServerOptions,
+)
 from azure.ai.agentserver.responses.models import CreateResponse, Item, OutputItem
+from azure.ai.agentserver.responses.streaming._checkpoint import ResponseCheckpointEvent
 from mcp import McpError
 from mcp.types import ErrorData
 from typing_extensions import Any
@@ -404,16 +411,29 @@ async def save_messages(
         with pytest.raises(RuntimeError, match="history provider"):
             ResponsesHostServer(agent)
 
+    def test_init_rejects_resilient_background_for_non_workflow_agent(self, tmp_path: Path) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        with pytest.raises(RuntimeError, match="resilient_background"):
+            ResponsesHostServer(
+                agent,
+                store=FileResponseStore(storage_dir=tmp_path),
+                options=ResponsesServerOptions(resilient_background=True),
+            )
+
     async def test_previous_response_requires_existing_agent_session(self) -> None:
         agent = _make_agent()
         server = _make_server(agent, session_store=SessionStore())
         request = CreateResponse(model="m", input="hi", previous_response_id="response-missing")
         context = ResponseContext(response_id="response-current", mode_flags=MagicMock())
 
-        handler = await server._handle_response(request, context, asyncio.Event())  # pyright: ignore[reportPrivateUsage]
+        handler = server._handle_response(request, context, asyncio.Event())  # pyright: ignore[reportPrivateUsage]
         events = [event async for event in handler]
 
-        failed_events = [event for event in events if event.get("type") == "response.failed"]
+        failed_events = [
+            event for event in events if isinstance(event, Mapping) and event.get("type") == "response.failed"
+        ]
         assert len(failed_events) == 1
         failed_event = cast(Mapping[str, Any], failed_events[0])
         response = cast(Mapping[str, Any], failed_event["response"])
@@ -464,6 +484,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
 
         provider = server._session_storage_provider  # pyright: ignore[reportPrivateUsage]
         assert provider is not None
+
         session_store = provider.get_store(config=server.config, platform_context=get_request_context())
         assert session_store is not None
         first_session = await session_store.get(first.json()["id"])
@@ -669,7 +690,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
         ):
             handler = cast(
                 AsyncGenerator[Any, None],
-                server._handle_inner_agent(request, context),  # pyright: ignore[reportPrivateUsage]
+                server._handle_response(request, context, asyncio.Event()),  # pyright: ignore[reportPrivateUsage]
             )
             await anext(handler)
             await anext(handler)
@@ -706,7 +727,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
         ):
             handler = cast(
                 AsyncGenerator[Any, None],
-                server._handle_inner_agent(request, context),  # pyright: ignore[reportPrivateUsage]
+                server._handle_response(request, context, asyncio.Event()),  # pyright: ignore[reportPrivateUsage]
             )
             await anext(handler)
             await anext(handler)
@@ -3518,13 +3539,16 @@ async def test_workflow_rejects_invalid_checkpoint_scope(
         with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])):
             events = [
                 event
-                async for event in server._handle_inner_workflow(  # pyright: ignore[reportPrivateUsage]
+                async for event in server._handle_response(  # pyright: ignore[reportPrivateUsage]
                     request,
                     ResponseContext(**context_kwargs),
+                    asyncio.Event(),
                 )
             ]
 
-        failed_events = [event for event in events if event.get("type") == "response.failed"]
+        failed_events = [
+            event for event in events if isinstance(event, Mapping) and event.get("type") == "response.failed"
+        ]
         assert len(failed_events) == 1
         failed_event = cast(Mapping[str, Any], failed_events[0])
         response = cast(Mapping[str, Any], failed_event["response"])
@@ -4386,3 +4410,36 @@ async def test_round_trip_approval_response_rejected(self) -> None:
 
 
 # endregion
+
+
+# region Resilient background checkpointing
+
+
+class TestResilientBackgroundCheckpointing:
+    """``ResponseEventStream.checkpoint()`` only persists when its returned event is ``yield``-ed, so
+    ``_handle_inner_workflow`` fetches the latest saved workflow checkpoint and yields it after every update
+    when resilient_background is enabled.
+    """
+
+    async def test_workflow_yields_checkpoint_event_when_resilient_background(self, tmp_path: Path) -> None:
+        workflow_agent = _build_text_workflow_agent("hello from workflow")
+        server = _make_server(
+            workflow_agent,
+            response_store=FileResponseStore(storage_dir=tmp_path),
+            options=ResponsesServerOptions(resilient_background=True),
+        )
+        request = CreateResponse(model="m", input="hi", background=True, stream=True, store=True)
+        context = ResponseContext(response_id="response-current", mode_flags=MagicMock())
+
+        events = [
+            event
+            async for event in server._handle_response(  # pyright: ignore[reportPrivateUsage]
+                request, context, asyncio.Event()
+            )
+        ]
+
+        checkpoint_events = [e for e in events if isinstance(e, ResponseCheckpointEvent)]
+        assert checkpoint_events, "expected at least one checkpoint event yielded for a resilient background run"
+
+
+# endregion

python/packages/foundry_hosting/tests/test_state_store.py

@@ -81,7 +81,7 @@ async def test_save_uses_context_scoped_store() -> None:
         result = await FoundryCheckpointStore("context-1", _platform_context()).save(checkpoint)
 
     assert result == "checkpoint-1"
-    get_or_create.assert_awaited_once_with("checkpoints/context-1", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("checkpoints/context-1", user_isolation=True)
     store.set_item.assert_awaited_once_with("checkpoint-1", checkpoint.to_dict(), call_id="call-1")
 
 
@@ -235,7 +235,7 @@ async def test_save_and_load_function_approval_request() -> None:
         loaded = await storage.load_approval_request("approval-1")
 
     assert get_or_create.await_count == 2
-    get_or_create.assert_awaited_with("function_approvals", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_with("function_approvals", user_isolation=True)
     store.create_item.assert_awaited_once_with("approval-1", request.to_dict(), call_id="call-1")
     store.get_item.assert_awaited_once_with("approval-1", call_id="call-1")
     assert loaded == request
@@ -311,7 +311,7 @@ async def test_set_agent_session_uses_scoped_store() -> None:
     ) as get_or_create:
         await FoundryAgentSessionStore(_platform_context()).set("storage-session-1", session)
 
-    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True)
     store.set_item.assert_awaited_once_with("storage-session-1", session.to_dict(), call_id="call-1")
 
 
@@ -330,7 +330,7 @@ async def test_get_agent_session_returns_deserialized_session() -> None:
     assert result is not None
     assert result.to_dict() == session.to_dict()
     assert result is not session
-    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True)
     store.get_item.assert_awaited_once_with("storage-session-1", call_id="call-1")
 
 

python/samples/04-hosting/foundry-hosted-agents/README.md

@@ -22,7 +22,9 @@ This directory contains samples that demonstrate how to use hosted [Agent Framew
 | 9 | [Foundry Memory](responses/foundry_memory/) | An agent with persistent semantic memory backed by a Microsoft Foundry Memory Store, using `FoundryMemoryProvider` to remember user facts across sessions. |
 | 10 | [Monty CodeAct](responses/monty_codeact/) | An agent with a Monty-backed CodeAct context provider, exposing a single `execute_code` tool that runs Python in a [pydantic-monty](https://github.com/pydantic/monty) interpreter and invokes typed host tools (`compute`, `fetch_data`) from inside the sandbox. Uses the beta `agent-framework-monty` package. |
 | 11 | [Foundry Toolbox MCP Skills](responses/foundry_toolbox_mcp_skills/) | An agent that discovers MCP-based skills attached to a Foundry Toolbox and serves them via `SkillsProvider(MCPSkillsSource(...))`, fetching `SKILL.md` bodies and supplementary resources on demand. |
-| 12 | [Using deployed agent](responses/using_deployed_agent.py) | Invoke an agent already deployed to Foundry using either a service-created or user-created hosted session, then delete the session after use. |
+| 13 | [Custom Storage](responses/custom_storage/) | An agent demonstrating how to implement a custom storage provider for agent sessions (in-memory and Cosmos DB). |
+| 14 | [Resilient Long-Running Workflow](responses/resilient_long_running_workflow/) | A long-running, crash-resilient workflow demonstrating how `resilient_background=True` lets a background response survive a hard crash of the server process and resume from its last checkpoint instead of restarting from scratch. |
+| 15 | [Using deployed agent](responses/using_deployed_agent.py) | Invoke an agent already deployed to Foundry using either a service-created or user-created hosted session, then delete the session after use. |
 
 ## Session Identifiers
 

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.dockerignore

@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+.env

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.env.example

@@ -0,0 +1,2 @@
+FOUNDRY_PROJECT_ENDPOINT="..."
+AZURE_AI_MODEL_DEPLOYMENT_NAME="..."

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY . user_agent/
+WORKDIR /app/user_agent
+
+RUN if [ -f requirements.txt ]; then \
+        pip install -r requirements.txt; \
+    else \
+        echo "No requirements.txt found"; \
+    fi
+
+EXPOSE 8088
+
+CMD ["python", "main.py"]

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/README.md

@@ -0,0 +1,112 @@
+# What this sample demonstrates
+
+A long-running, crash-resilient [Agent Framework](https://github.com/microsoft/agent-framework) workflow
+hosted using the **Responses protocol**. The workflow extracts a target number from the user's message and
+then counts down from it one step per second, demonstrating how `resilient_background=True` lets a
+background response survive a hard crash of the server process and resume from its last checkpoint instead
+of restarting from scratch.
+
+## How It Works
+
+### Workflow
+
+The workflow has three executors (see [main.py](main.py)):
+
+- **`StartExecutor`** uses a `FoundryChatClient`-backed agent to extract a positive integer target from the
+  user's message. If no valid target is found, the workflow yields an error message instead of counting down.
+- **`CountdownExecutor`** decrements the target through a self-loop, sleeping for a second and yielding an
+  output on each tick, to simulate a long-running operation.
+- **`complete`** yields the workflow's final `"Countdown complete."` output once the countdown reaches zero.
+
+### Agent Hosting
+
+The workflow is hosted as an agent using the [Agent Framework](https://github.com/microsoft/agent-framework)
+`ResponsesHostServer`, which provisions a REST API endpoint compatible with the OpenAI Responses protocol.
+Setting `resilient_background=True` in `ResponsesServerOptions` enables the framework to checkpoint the
+workflow's progress and durably persist streamed output, so a background response can be recovered and
+resumed after a crash (see "Testing resiliency" below).
+
+## Running the Agent Host
+
+Follow the instructions in the [Running the Agent Host Locally](../../README.md#running-the-agent-host-locally) section of the README in the parent directory to run the agent host.
+
+## Interacting with the agent
+
+> Depending on how you run the agent host, you can invoke the agent using `curl` (`Invoke-WebRequest` in PowerShell) or `azd`. Please refer to the [parent README](../../README.md) for more details. Use this README for sample queries you can send to the agent.
+
+Send a POST request to the server with a JSON body containing an `"input"` field with a positive integer target to count down from. For example:
+
+```bash
+curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" -d '{"input": "Count down from 5"}'
+```
+
+The server will respond with a JSON object containing the response output (one item per countdown step) and a response ID. You can use this response ID to continue the conversation in subsequent requests.
+
+
+## Testing resiliency (crash recovery)
+
+This sample enables `resilient_background=True`, so a long-running countdown survives a hard crash of the
+server process and resumes from its last checkpoint instead of restarting from scratch. Locally, the
+server persists responses, streams, and checkpoints under `${AGENTSERVER_STATE_ROOT:-~/.agentserver}/`, so
+this state survives a process restart as long as you run from the same working directory.
+
+On startup, the server's task manager scans that persisted state for any tasks that were still in flight
+when the process died and automatically reclaims and resumes each one from its last checkpoint -- this
+happens for every incomplete resilient background response, not just the one a client happens to reconnect
+to. This is why a stale response from an earlier run can still be found "recovering" in the server logs
+long after you've moved on to a new test: every restart re-triggers the same scan, so the stale task keeps
+getting resumed until it either reaches a terminal state or the persisted state directory is cleared (as
+`verify_resiliency.py` does).
+
+### Automated
+
+[verify_resiliency.py](verify_resiliency.py) runs the whole scenario end to end: it clears any leftover
+`${AGENTSERVER_STATE_ROOT:-~/.agentserver}` state from a previous run, starts the server, kicks off a
+background+streaming countdown, force-kills the server once half the countdown has completed, restarts the
+server, and asserts the recovered response completes with the exact expected output (no lost or duplicated
+steps). Progress is printed as each countdown item completes, both before and after the crash, by reading
+the response's own `stream=true` SSE feed -- a plain (non-streaming) `GET` only ever reflects the response's
+initial or terminal snapshot, never anything in between.
+
+```bash
+python verify_resiliency.py --target 20
+```
+
+### Manual
+
+To exercise crash recovery by hand:
+
+1. Start the server, then kick off a long background+streaming countdown and note the response `id` from the
+   first `response.created` event:
+
+   ```bash
+   curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+     -d '{"input": "Count down from 200", "stream": true, "store": true, "background": true}'
+   ```
+
+2. While the countdown is still running, kill the server process abruptly — use `kill -9 <pid>`
+   (`Stop-Process -Id <pid> -Force` on Windows), not `Ctrl+C`. A `Ctrl+C` triggers a graceful shutdown, which
+   is handled differently than a crash; a hard kill is required to exercise crash recovery. The sample server
+   prints its PID (`PID: <pid>`) on startup so you don't need to look it up separately.
+
+3. Restart the server (`python main.py`) from the same working directory.
+
+4. Reconnect to the response to observe recovery:
+
+   ```bash
+   curl "http://localhost:8088/responses/REPLACE_WITH_RESPONSE_ID?stream=true"
+   ```
+
+   The recovered stream emits a fresh `response.in_progress` event first, then resumes the countdown from
+   where it left off — the output already produced before the crash is neither lost nor duplicated.
+   Alternatively, poll `GET /responses/REPLACE_WITH_RESPONSE_ID` (without `stream`) until `status` is
+   `completed` and inspect the `output` array for a contiguous, non-duplicated sequence.
+
+> **Windows note:** the local stream store falls back to a plain lock *file* (no `fcntl`), which isn't
+> cleaned up when the process is force-killed. If restart fails with `another process holds the lock-file
+> on ...jsonl`, delete the stale `<response-id>.jsonl.lock` file under
+> `%USERPROFILE%\.agentserver\streams\` before restarting the server.
+
+## Deploying the Agent to Foundry
+
+To host the agent on Foundry, follow the instructions in the [Deploying the Agent to Foundry](../../README.md#deploying-the-agent-to-foundry) section of the README in the parent directory.

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.manifest.yaml

@@ -0,0 +1,25 @@
+name: agent-framework-resilient-long-running-workflow
+description: >
+  A hosted agent that demonstrates a resilient long-running workflow using the Responses protocol.
+metadata:
+  tags:
+    - Agent Framework
+    - AI Agent Hosting
+    - Azure AI AgentServer
+    - Responses Protocol
+    - Streaming
+    - Workflows
+    - Resilience
+template:
+  name: agent-framework-resilient-long-running-workflow
+  kind: hosted
+  protocols:
+    - protocol: responses
+      version: 2.0.0
+  environment_variables:
+    - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+      value: "{{AZURE_AI_MODEL_DEPLOYMENT_NAME}}"
+resources:
+  - kind: model
+    id: gpt-4.1-mini
+    name: AZURE_AI_MODEL_DEPLOYMENT_NAME

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.yaml

@@ -0,0 +1,12 @@
+# yaml-language-server: $schema=https://raw.githubusercontent.com/microsoft/AgentSchema/refs/heads/main/schemas/v1.0/ContainerAgent.yaml
+kind: hosted
+name: agent-framework-resilient-long-running-workflow
+protocols:
+  - protocol: responses
+    version: 2.0.0
+resources:
+  cpu: "0.25"
+  memory: "0.5Gi"
+environment_variables:
+  - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+    value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}