Update type ignore statements

_resumed_from_checkpoint was only cleared on the success path of run_until_convergence, so a failure during a resumed run (e.g. executor failure) left it True. The next fresh run then skipped the superstep-0 checkpoint and parented later checkpoints to the stale resume point. Move the reset into a finally. Add a regression test that fails a resumed run via an executor error and asserts the next fresh run creates the superstep-0 checkpoint.

dotnet/nuget/nuget-package.props

@@ -3,7 +3,7 @@
     <!-- Central version prefix - applies to all nuget packages. -->
     <VersionPrefix>1.11.0</VersionPrefix>
     <RCNumber>1</RCNumber>
-    <DateSuffix>260622</DateSuffix>
+    <DateSuffix>260623</DateSuffix>
     <PackageVersion Condition="'$(IsReleaseCandidate)' == 'true'">$(VersionPrefix)-rc$(RCNumber)</PackageVersion>
     <PackageVersion Condition="'$(IsReleaseCandidate)' != 'true' AND '$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix).$(DateSuffix).1</PackageVersion>
     <PackageVersion Condition="'$(IsReleaseCandidate)' != 'true' AND '$(VersionSuffix)' == ''">$(VersionPrefix)-preview.$(DateSuffix).1</PackageVersion>

python/PACKAGE_STATUS.md

@@ -34,6 +34,8 @@ Status is grouped into these buckets:
 | `agent-framework-foundry-local` | `python/packages/foundry_local` | `beta` |
 | `agent-framework-gemini` | `python/packages/gemini` | `alpha` |
 | `agent-framework-github-copilot` | `python/packages/github_copilot` | `rc` |
+| `agent-framework-hosting` | `python/packages/hosting` | `alpha` |
+| `agent-framework-hosting-responses` | `python/packages/hosting-responses` | `alpha` |
 | `agent-framework-hyperlight` | `python/packages/hyperlight` | `beta` |
 | `agent-framework-lab` | `python/packages/lab` | `beta` |
 | `agent-framework-mem0` | `python/packages/mem0` | `beta` |

python/packages/core/AGENTS.md

@@ -80,8 +80,9 @@ agent_framework/
 
 - **`MCPTool`** - Base wrapper that owns the MCP `ClientSession` and exposes the remote server's tools as `FunctionTool`s.
 - **`MCPStdioTool`** / **`MCPStreamableHTTPTool`** / **`MCPWebsocketTool`** - Transport-specific subclasses.
-- **Argument allowlist (`_prepare_call_kwargs`)** - Before each `tools/call`, kwargs are filtered to an **allowlist** built from the tool's declared parameters (`inputSchema.properties`) plus any user-configured extras. Framework runtime kwargs injected through the function-invocation pipeline (e.g. `thread`, `conversation_id`, `chat_options`, `options`, `response_format`) are stripped by default rather than forwarded. A tool that declares no usable `properties` (including schemas with `additionalProperties: true`) forwards only the configured extras. The `_MCP_FRAMEWORK_DENYLIST` is a safety net for framework-named params a server *declares* in its schema (those are dropped); names explicitly opted in via `additional_tool_argument_names` always win. The reserved `_meta` key is extracted as MCP request metadata, never forwarded as an argument.
-- **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins.
+- **Argument allowlist (`_prepare_call_kwargs`)** - Before each `tools/call`, kwargs are filtered to an **allowlist** built from the tool's declared parameters (`inputSchema.properties`) plus any user-configured extras. Framework runtime kwargs injected through the function-invocation pipeline (e.g. `thread`, `conversation_id`, `chat_options`, `options`, `response_format`) are stripped by default rather than forwarded. A tool that declares no usable `properties` (including schemas with `additionalProperties: true`) forwards only the configured extras. The `_MCP_FRAMEWORK_DENYLIST` is a safety net for framework-named params a server *declares* in its schema (those are dropped); names explicitly opted in via `additional_tool_argument_names` always win. The reserved `_meta` key is never forwarded as an argument; trusted caller/runtime `_meta` is validated as MCP request metadata, model-supplied `_meta` is discarded in generated MCP functions, and metadata precedence is caller/runtime < OpenTelemetry < tools/list metadata.
+- **`allowed_tools`** (constructor arg on all `MCPTool` subclasses) - Restricts exposed MCP tools by raw remote MCP tool identity. Prefixed local names remain accepted only when the raw remote name already matches its normalized form; normalized/local aliases do not authorize a different raw remote name. If multiple raw remote tool names map to the same local function name, tool loading raises `ToolExecutionException` instead of first-one-wins shadowing.
+- **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a normal forwarded argument name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins; `_meta` is the exception and only trusted runtime/caller metadata is used.
 - **Sampling guardrails** (`sampling_callback`) - Passing `client=` advertises `SamplingCapability` so the server can send `sampling/createMessage`. Because remote servers are untrusted (confused-deputy risk), the default `sampling_callback` is **deny-by-default** and applies, in order: a per-session rate limit (`sampling_max_requests`, default `_DEFAULT_SAMPLING_MAX_REQUESTS`), an approval gate (`sampling_approval_callback`), and a `maxTokens` cap (`sampling_max_tokens`, default `_DEFAULT_SAMPLING_MAX_TOKENS`). The approval callback (constructor arg on all subclasses; exported type alias `SamplingApprovalCallback`) receives the raw `CreateMessageRequestParams`, may be sync or async, and must return truthy to approve. When it is `None` (the default) every sampling request is denied; pass `lambda params: True` to restore legacy auto-approve as an explicit opt-in. Requests and denials are logged at WARNING (content is not logged). The per-session counter resets in `_reset_session_state`.
 - **`MCPTaskOptions`** (experimental, `MCP_LONG_RUNNING_TASKS` feature, **frozen**) - Per-tool-instance options controlling the SEP-2663 long-running task lifecycle. When the server advertises a tool with `execution.taskSupport == "required"`, `MCPTool.call_tool` transparently routes through `call_tool_as_task`, which sends an augmented `tools/call`, polls `tasks/get` until terminal, and reinterprets `tasks/result` as a normal `CallToolResult`. Instances are immutable; replace via `MCPTool.task_options = MCPTaskOptions(...)`. Fields:
   - `default_ttl: timedelta | None` — forwarded to the server as `params.task.ttl` (milliseconds). When `None`, the server's default applies.

python/packages/core/agent_framework/_mcp.py

@@ -74,6 +74,11 @@ _MCP_NORMALIZED_NAME_KEY = "_mcp_normalized_name"
 # Reserved key in an ``additional_tool_argument_names`` mapping that applies its
 # values to every tool on the server rather than a single named tool.
 _MCP_GLOBAL_EXTRA_ARGS_KEY = "*"
+_MCP_META_LABEL_PATTERN = r"[A-Za-z](?:[A-Za-z0-9-]*[A-Za-z0-9])?"
+_MCP_META_KEY_PATTERN = re.compile(
+    rf"^(?:(?:{_MCP_META_LABEL_PATTERN})(?:\.{_MCP_META_LABEL_PATTERN})*/)?"
+    r"[A-Za-z0-9](?:[A-Za-z0-9_.-]*[A-Za-z0-9])?$"
+)
 # Framework kwargs that flow through the function-invocation pipeline (via
 # ``FunctionInvocationContext.kwargs``) but must never be forwarded to an MCP
 # server: they are internal objects that the MCP SDK cannot serialize. They are
@@ -205,7 +210,42 @@ def _normalize_additional_tool_argument_names(
     return set(additional_tool_argument_names), {}
 
 
-def _inject_otel_into_mcp_meta(meta: dict[str, Any] | None = None) -> dict[str, Any] | None:
+def _mcp_config_candidate_names(*, local_name: str, normalized_name: str, remote_name: str) -> tuple[str, ...]:
+    """Return safe configuration names for MCP allow/approval matching."""
+    names = [remote_name]
+    if normalized_name == remote_name and local_name != remote_name:
+        names.append(local_name)
+    return tuple(names)
+
+
+def _validate_mcp_meta_key(key: str) -> None:
+    """Validate an MCP ``_meta`` key against the 2025-06-18 key-name format."""
+    if not _MCP_META_KEY_PATTERN.fullmatch(key):
+        raise ToolExecutionException(f"Invalid MCP _meta key name: {key!r}.")
+
+
+def _validate_mcp_meta(raw_meta: object | None) -> dict[str, Any] | None:
+    """Validate and copy MCP request metadata."""
+    if raw_meta is None:
+        return None
+    if not isinstance(raw_meta, dict):
+        raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
+
+    raw_meta_dict = cast(Mapping[object, Any], raw_meta)
+    meta: dict[str, Any] = {}
+    for key, value in raw_meta_dict.items():
+        if not isinstance(key, str):
+            raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
+        _validate_mcp_meta_key(key)
+        meta[key] = value
+    return meta
+
+
+def _inject_otel_into_mcp_meta(
+    meta: dict[str, Any] | None = None,
+    *,
+    overwrite: bool = False,
+) -> dict[str, Any] | None:
     """Inject OpenTelemetry trace context into MCP request _meta via the global propagator(s)."""
     carrier: dict[str, str] = {}
     propagate.inject(carrier)
@@ -215,7 +255,8 @@ def _inject_otel_into_mcp_meta(meta: dict[str, Any] | None = None) -> dict[str,
     if meta is None:
         meta = {}
     for key, value in carrier.items():
-        if key not in meta:
+        _validate_mcp_meta_key(key)
+        if overwrite or key not in meta:
             meta[key] = value
 
     return meta
@@ -381,7 +422,9 @@ class MCPTool:
             approval_mode: Whether approval is required to run tools.
             allowed_tools: Optional allow-list of MCP tool names to expose as functions.
                 ``None`` (the default) exposes every tool advertised by the MCP server.
-                A non-empty collection exposes only the tools whose names appear in it.
+                A non-empty collection exposes only the raw remote tools whose names appear in it. For
+                compatibility, the prefixed local function name is also accepted when the raw remote name already
+                matches its normalized form; normalized aliases do not authorize a different raw remote tool.
                 An empty collection (``[]``) exposes no tools — if you simply want to
                 disable tool execution, prefer ``load_tools=False`` instead. ``[]`` is
                 useful as a runtime guard or when you want to load tool metadata for
@@ -753,11 +796,14 @@ class MCPTool:
             additional_properties = func.additional_properties or {}
             normalized_name = additional_properties.get(_MCP_NORMALIZED_NAME_KEY)
             remote_name = additional_properties.get(_MCP_REMOTE_NAME_KEY)
-            if (
-                func.name in allowed_names
-                or (isinstance(normalized_name, str) and normalized_name in allowed_names)
-                or (isinstance(remote_name, str) and remote_name in allowed_names)
-            ):
+            if not isinstance(normalized_name, str) or not isinstance(remote_name, str):
+                continue
+            candidate_names = _mcp_config_candidate_names(
+                local_name=func.name,
+                normalized_name=normalized_name,
+                remote_name=remote_name,
+            )
+            if any(name in allowed_names for name in candidate_names):
                 filtered_functions.append(func)
         return filtered_functions
 
@@ -1381,7 +1427,13 @@ class MCPTool:
                     continue
 
                 input_model = _get_input_model_from_mcp_prompt(prompt)
-                approval_mode = self._determine_approval_mode(local_name, normalized_name, prompt.name)
+                approval_mode = self._determine_approval_mode(
+                    *_mcp_config_candidate_names(
+                        local_name=local_name,
+                        normalized_name=normalized_name,
+                        remote_name=prompt.name,
+                    )
+                )
                 func: FunctionTool = FunctionTool(
                     func=partial(self.get_prompt, prompt.name),
                     name=local_name,
@@ -1422,7 +1474,11 @@ class MCPTool:
             return
 
         # Track existing function names to prevent duplicates
-        existing_names = {func.name for func in self._functions}
+        existing_remote_by_local: dict[str, str] = {}
+        for func in self._functions:
+            remote_name = (func.additional_properties or {}).get(_MCP_REMOTE_NAME_KEY)
+            if isinstance(remote_name, str):
+                existing_remote_by_local[func.name] = remote_name
         tool_call_meta_by_name: dict[str, dict[str, Any]] = {}
         tool_task_support_by_name: dict[str, str] = {}
         tool_param_names_by_name: dict[str, set[str]] = {}
@@ -1462,7 +1518,7 @@ class MCPTool:
 
             for tool in tool_list.tools:
                 if tool.meta is not None:
-                    tool_call_meta_by_name[tool.name] = dict(tool.meta)
+                    tool_call_meta_by_name[tool.name] = _validate_mcp_meta(tool.meta) or {}
 
                 task_support = getattr(getattr(tool, "execution", None), "taskSupport", None)
                 if task_support is not None:
@@ -1490,10 +1546,24 @@ class MCPTool:
                 local_name = _build_prefixed_mcp_name(normalized_name, self.tool_name_prefix)
 
                 # Skip if already loaded
-                if local_name in existing_names:
+                if local_name in existing_remote_by_local:
+                    if existing_remote_by_local.get(local_name) != tool.name:
+                        raise ToolExecutionException(
+                            "MCP server advertised multiple tools that map to the same local function name: "
+                            f"{existing_remote_by_local[local_name]!r} and {tool.name!r} both map to "
+                            f"{local_name!r}."
+                        )
                     continue
 
-                approval_mode = self._determine_approval_mode(local_name, normalized_name, tool.name)
+                existing_remote_by_local[local_name] = tool.name
+
+                approval_mode = self._determine_approval_mode(
+                    *_mcp_config_candidate_names(
+                        local_name=local_name,
+                        normalized_name=normalized_name,
+                        remote_name=tool.name,
+                    )
+                )
 
                 async def _call_tool_with_runtime_kwargs(
                     ctx: FunctionInvocationContext,
@@ -1501,8 +1571,13 @@ class MCPTool:
                     _remote_tool_name: str = tool.name,
                     **kwargs: Any,
                 ) -> str | list[Content]:
+                    trusted_meta = ctx.kwargs.get("_meta")
                     call_kwargs = dict(ctx.kwargs)
                     call_kwargs.update(kwargs)
+                    if trusted_meta is not None:
+                        call_kwargs["_meta"] = trusted_meta
+                    else:
+                        call_kwargs.pop("_meta", None)
                     return await self.call_tool(_remote_tool_name, **call_kwargs)
 
                 # Create FunctionTools out of each tool
@@ -1518,7 +1593,6 @@ class MCPTool:
                     },
                 )
                 self._functions.append(func)
-                existing_names.add(local_name)
 
             # Check if there are more pages
             if not tool_list.nextCursor:
@@ -1636,8 +1710,8 @@ class MCPTool:
         Keyword Args:
             _meta: Optional ``dict[str, Any]`` of MCP request metadata. This reserved key is passed as the
                 ``meta`` parameter of the underlying ``session.call_tool`` call rather than as a tool argument.
-                User-supplied keys override metadata from ``tools/list``; OpenTelemetry propagation fills in
-                non-conflicting keys.
+                OpenTelemetry propagation overrides caller-supplied keys, and metadata from ``tools/list``
+                overrides both.
             kwargs: Remaining arguments to pass to the tool.
 
         Returns:
@@ -1746,17 +1820,7 @@ class MCPTool:
         self, tool_name: str, kwargs: dict[str, Any]
     ) -> tuple[dict[str, Any], dict[str, Any] | None]:
         """Filter kwargs down to the tool's arguments and build the merged MCP request metadata."""
-        raw_user_meta: object | None = kwargs.get("_meta")
-        user_meta: dict[str, Any] | None = None
-        if raw_user_meta is not None and not isinstance(raw_user_meta, dict):
-            raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
-        if isinstance(raw_user_meta, dict):
-            raw_user_meta_dict = cast(Mapping[object, object], raw_user_meta)
-            user_meta = {}
-            for key, value in raw_user_meta_dict.items():
-                if not isinstance(key, str):
-                    raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
-                user_meta[key] = value
+        user_meta = _validate_mcp_meta(kwargs.get("_meta"))
 
         # Allowlist: forward only the tool's declared parameters (from inputSchema.properties)
         # plus any user-configured extra argument names. Everything else - notably the
@@ -1783,12 +1847,12 @@ class MCPTool:
         }
 
         # Some MCP proxies require their tools/list metadata to be echoed on tools/call.
-        tool_meta = self._tool_call_meta_by_name.get(tool_name)
-        request_meta = dict(tool_meta) if tool_meta is not None else None
-        if user_meta is not None:
-            request_meta = {**(request_meta or {}), **user_meta}
-        meta = _inject_otel_into_mcp_meta(request_meta)
-        return filtered_kwargs, meta
+        request_meta = dict(user_meta) if user_meta is not None else None
+        request_meta = _inject_otel_into_mcp_meta(request_meta, overwrite=True)
+        tool_meta = _validate_mcp_meta(self._tool_call_meta_by_name.get(tool_name))
+        if tool_meta is not None:
+            request_meta = {**(request_meta or {}), **tool_meta}
+        return filtered_kwargs, request_meta
 
     async def call_tool_as_task(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
         """Call an MCP tool via the long-running task lifecycle (SEP-2663).

python/packages/core/agent_framework/_tools.py

@@ -1004,39 +1004,6 @@ def normalize_tools(
     return normalized
 
 
-def _tools_to_dict(  # pyright: ignore[reportUnusedFunction]
-    tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None,
-) -> list[str | dict[str, Any]] | None:
-    """Parse the tools to a dict.
-
-    Args:
-        tools: The tools to parse. Can be a single tool or a sequence of tools.
-
-    Returns:
-        A list of tool specifications as dictionaries, or None if no tools provided.
-    """
-    normalized_tools = normalize_tools(tools)
-    if not normalized_tools:
-        return None
-
-    results: list[str | dict[str, Any]] = []
-    for tool_item in normalized_tools:
-        if isinstance(tool_item, FunctionTool):
-            results.append(tool_item.to_json_schema_spec())
-            continue
-        if isinstance(tool_item, BaseModel):
-            results.append(tool_item.model_dump(exclude_none=True))
-            continue
-        if isinstance(tool_item, SerializationMixin):
-            results.append(tool_item.to_dict())
-            continue
-        if isinstance(tool_item, dict):
-            results.append(tool_item)  # type: ignore[reportUnknownArgumentType]
-            continue
-        logger.warning("Can't parse tool.")
-    return results
-
-
 # region AI Function Decorator
 
 

python/packages/core/agent_framework/_types.py

@@ -2983,6 +2983,8 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         self._inner_stream_source: ResponseStream[Any, Any] | Awaitable[ResponseStream[Any, Any]] | None = None
         self._wrap_inner: bool = False
         self._map_update: Callable[[Any], UpdateT | Awaitable[UpdateT]] | None = None
+        self._flat_map_update: Callable[[Any], Iterable[UpdateT] | Awaitable[Iterable[UpdateT]]] | None = None
+        self._pending_mapped_updates: list[UpdateT] = []
         self._pull_context_manager_factories: list[Callable[[], contextlib.AbstractContextManager[Any]]] = []
 
     def map(
@@ -3029,6 +3031,23 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         stream._map_update = transform
         return stream
 
+    def flat_map(
+        self,
+        transform: Callable[[UpdateT], Iterable[OuterUpdateT] | Awaitable[Iterable[OuterUpdateT]]],
+        finalizer: Callable[[Sequence[OuterUpdateT]], OuterFinalT | Awaitable[OuterFinalT]],
+    ) -> ResponseStream[OuterUpdateT, OuterFinalT]:
+        """Create a new stream that transforms each update into zero or more updates.
+
+        Like :meth:`map`, the returned stream delegates iteration to this stream,
+        preserving single consumption and inner finalization/result hooks. Use this
+        when one upstream update naturally expands into multiple wire-protocol events.
+        """
+        stream: ResponseStream[OuterUpdateT, OuterFinalT] = ResponseStream(self, finalizer=finalizer)
+        stream._inner_stream_source = self
+        stream._wrap_inner = True
+        stream._flat_map_update = transform
+        return stream
+
     def with_finalizer(
         self,
         finalizer: Callable[[Sequence[UpdateT]], OuterFinalT | Awaitable[OuterFinalT]],
@@ -3101,35 +3120,7 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
     def __aiter__(self) -> ResponseStream[UpdateT, FinalT]:
         return self
 
-    async def __anext__(self) -> UpdateT:
-        try:
-            with contextlib.ExitStack() as stack:
-                for factory in self._pull_context_manager_factories:
-                    stack.enter_context(factory())
-                # Resolve the underlying stream inside the pull contexts so that any
-                # spans/contexts created during stream resolution (e.g. inner chat
-                # completion spans created on the first pull of a wrapped agent stream)
-                # inherit the active context (e.g. an outer agent invoke span).
-                if self._iterator is None:
-                    stream = await self._get_stream()
-                    self._iterator = stream.__aiter__()
-                update: UpdateT = await self._iterator.__anext__()
-        except StopAsyncIteration:
-            self._consumed = True
-            await self._run_cleanup_hooks()
-            await self.get_final_response()
-            raise
-        except Exception as exc:
-            self._stream_error = exc
-            try:
-                await self._run_cleanup_hooks()
-            finally:
-                self._stream_error = None
-            raise
-        if self._map_update is not None:
-            update = self._map_update(update)  # type: ignore[assignment]
-            if isawaitable(update):
-                update = await update
+    async def _record_update(self, update: UpdateT) -> UpdateT:
         self._updates.append(update)
         for hook in self._transform_hooks:
             hooked = hook(update)
@@ -3139,6 +3130,47 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
                 update = cast(UpdateT, hooked)
         return update
 
+    async def __anext__(self) -> UpdateT:
+        while True:
+            if self._pending_mapped_updates:
+                return await self._record_update(self._pending_mapped_updates.pop(0))
+
+            try:
+                with contextlib.ExitStack() as stack:
+                    for factory in self._pull_context_manager_factories:
+                        stack.enter_context(factory())
+                    # Resolve the underlying stream inside the pull contexts so that any
+                    # spans/contexts created during stream resolution (e.g. inner chat
+                    # completion spans created on the first pull of a wrapped agent stream)
+                    # inherit the active context (e.g. an outer agent invoke span).
+                    if self._iterator is None:
+                        stream = await self._get_stream()
+                        self._iterator = stream.__aiter__()
+                    update: UpdateT = await self._iterator.__anext__()
+            except StopAsyncIteration:
+                self._consumed = True
+                await self._run_cleanup_hooks()
+                await self.get_final_response()
+                raise
+            except Exception as exc:
+                self._stream_error = exc
+                try:
+                    await self._run_cleanup_hooks()
+                finally:
+                    self._stream_error = None
+                raise
+            if self._flat_map_update is not None:
+                mapped_updates = self._flat_map_update(update)
+                if isawaitable(mapped_updates):
+                    mapped_updates = await mapped_updates
+                self._pending_mapped_updates.extend(mapped_updates)
+                continue
+            if self._map_update is not None:
+                update = self._map_update(update)  # type: ignore[assignment]
+                if isawaitable(update):
+                    update = await update
+            return await self._record_update(update)
+
     async def _resolve_stream_with_pull_contexts(self) -> AsyncIterable[UpdateT]:
         """Resolve the underlying stream while activating any registered pull context managers.
 

python/packages/core/agent_framework/_workflows/_runner.py

@@ -11,7 +11,6 @@ from typing import Any
 from ..exceptions import (
     WorkflowCheckpointException,
     WorkflowConvergenceException,
-    WorkflowException,
 )
 from ._checkpoint import CheckpointID, CheckpointStorage, WorkflowCheckpoint
 from ._const import EXECUTOR_STATE_KEY
@@ -105,97 +104,84 @@ class Runner:
         """
         self._iteration = 0
 
-    def reset_runtime_state(
-        self,
-        *,
-        iteration: int = 0,
-        previous_checkpoint_id: CheckpointID | None = None,
-        resumed_from_checkpoint: bool = False,
-    ) -> None:
-        """Reset runner runtime bookkeeping to a known baseline.
-
-        Args:
-            iteration: Iteration value to restore.
-            previous_checkpoint_id: Checkpoint parent pointer for subsequent saves.
-            resumed_from_checkpoint: Whether to treat next run as resumed.
-        """
-        self._iteration = iteration
-        self._previous_checkpoint_id = previous_checkpoint_id
-        self._resumed_from_checkpoint = resumed_from_checkpoint
-
     async def run_until_convergence(self) -> AsyncGenerator[WorkflowEvent, None]:
         """Run the workflow until no more messages are sent."""
-        # Emit any events already produced prior to entering loop
-        if await self._ctx.has_events():
-            logger.info("Yielding pre-loop events")
-            for event in await self._ctx.drain_events():
-                yield event
-
-        # Create a checkpoint before a run starts. Checkpoints are usually considered to be created at the
-        # end of an iteration, we can think of this checkpoint as being created at the end of "superstep 0"
-        # which captures the states after which the start executor has run. Note that we execute the start
-        # executor outside of the main iteration loop.
-        if await self._ctx.has_messages() and not self._resumed_from_checkpoint:
-            await self.create_checkpoint_if_enabled()
-
-        while self._iteration < self._max_iterations:
-            logger.info(f"Starting superstep {self._iteration + 1}")
-            yield WorkflowEvent.superstep_started(iteration=self._iteration + 1)
-
-            # Run iteration concurrently with live event streaming: we poll
-            # for new events while the iteration coroutine progresses.
-            iteration_task = asyncio.create_task(self._run_iteration())
-            try:
-                while not iteration_task.done():
-                    try:
-                        # Wait briefly for any new event; timeout allows progress checks
-                        event = await asyncio.wait_for(self._ctx.next_event(), timeout=0.05)
-                        yield event
-                    except asyncio.TimeoutError:
-                        # Periodically continue to let iteration advance
-                        continue
-            except asyncio.CancelledError:
-                # Propagate cancellation to the iteration task to avoid orphaned work
-                iteration_task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await iteration_task
-                raise
-
-            # Propagate errors from iteration, but first surface any pending events
-            try:
-                await iteration_task
-            except Exception:
-                # Make sure failure-related events (like ExecutorFailedEvent) are surfaced
-                if await self._ctx.has_events():
-                    for event in await self._ctx.drain_events():
-                        yield event
-                raise
-            self._iteration += 1
-
-            # Drain any straggler events emitted at tail end
+        try:
+            # Emit any events already produced prior to entering loop
             if await self._ctx.has_events():
+                logger.info("Yielding pre-loop events")
                 for event in await self._ctx.drain_events():
                     yield event
 
-            logger.info(f"Completed superstep {self._iteration}")
+            # Create a checkpoint before a run starts. Checkpoints are usually considered to be created at the
+            # end of an iteration, we can think of this checkpoint as being created at the end of "superstep 0"
+            # which captures the states after which the start executor has run. Note that we execute the start
+            # executor outside of the main iteration loop.
+            if await self._ctx.has_messages() and self._iteration == 0 and not self._resumed_from_checkpoint:
+                await self.create_checkpoint_if_enabled()
 
-            # Commit pending state changes at superstep boundary
-            self._state.commit()
+            while self._iteration < self._max_iterations:
+                logger.info(f"Starting superstep {self._iteration + 1}")
+                yield WorkflowEvent.superstep_started(iteration=self._iteration + 1)
 
-            # Create checkpoint after each superstep iteration
-            await self.create_checkpoint_if_enabled()
+                # Run iteration concurrently with live event streaming: we poll
+                # for new events while the iteration coroutine progresses.
+                iteration_task = asyncio.create_task(self._run_iteration())
+                try:
+                    while not iteration_task.done():
+                        try:
+                            # Wait briefly for any new event; timeout allows progress checks
+                            event = await asyncio.wait_for(self._ctx.next_event(), timeout=0.05)
+                            yield event
+                        except asyncio.TimeoutError:
+                            # Periodically continue to let iteration advance
+                            continue
+                except asyncio.CancelledError:
+                    # Propagate cancellation to the iteration task to avoid orphaned work
+                    iteration_task.cancel()
+                    with contextlib.suppress(asyncio.CancelledError):
+                        await iteration_task
+                    raise
 
-            yield WorkflowEvent.superstep_completed(iteration=self._iteration)
+                # Propagate errors from iteration, but first surface any pending events
+                try:
+                    await iteration_task
+                except Exception:
+                    # Make sure failure-related events (like ExecutorFailedEvent) are surfaced
+                    if await self._ctx.has_events():
+                        for event in await self._ctx.drain_events():
+                            yield event
+                    raise
+                self._iteration += 1
 
-            # Check for convergence: no more messages to process
-            if not await self._ctx.has_messages():
-                break
+                # Drain any straggler events emitted at tail end
+                if await self._ctx.has_events():
+                    for event in await self._ctx.drain_events():
+                        yield event
 
-        logger.info(f"Workflow completed after {self._iteration} supersteps")
-        self._resumed_from_checkpoint = False  # Reset resume flag for next run
+                logger.info(f"Completed superstep {self._iteration}")
 
-        if self._iteration >= self._max_iterations and await self._ctx.has_messages():
-            raise WorkflowConvergenceException(f"Runner did not converge after {self._max_iterations} iterations.")
+                # Commit pending state changes at superstep boundary
+                self._state.commit()
+
+                # Create checkpoint after each superstep iteration
+                await self.create_checkpoint_if_enabled()
+
+                yield WorkflowEvent.superstep_completed(iteration=self._iteration)
+
+                # Check for convergence: no more messages to process
+                if not await self._ctx.has_messages():
+                    break
+
+            logger.info(f"Workflow completed after {self._iteration} supersteps")
+
+            if self._iteration >= self._max_iterations and await self._ctx.has_messages():
+                raise WorkflowConvergenceException(f"Runner did not converge after {self._max_iterations} iterations.")
+        finally:
+            # Reset the resume flag so stale resume state never leaks into the next run on this
+            # instance - even if convergence raised before completing (e.g. an executor failure
+            # during a resumed run).
+            self._resumed_from_checkpoint = False
 
     async def _run_iteration(self) -> None:
         """Run a single iteration of the workflow.
@@ -258,46 +244,6 @@ class Runner:
         await self._save_executor_states()
         self._state.commit()
 
-    async def capture_checkpoint_object(self, *, metadata: dict[str, Any] | None = None) -> WorkflowCheckpoint:
-        """Capture the current runner state as an in-memory checkpoint object.
-
-        Persists executor snapshots into committed state and builds a
-        ``WorkflowCheckpoint`` from the current committed state. The checkpoint is
-        not written to any storage backend; the caller owns its lifetime (for
-        example, the workflow's captured initial checkpoint used by reset).
-
-        This is only valid when the runner is quiescent: it rejects capture when
-        in-flight executor messages or pending request_info events are present,
-        since those represent mid-run state that would not form a clean baseline.
-
-        Args:
-            metadata: Optional metadata to attach to the checkpoint.
-
-        Returns:
-            A ``WorkflowCheckpoint`` snapshot of the current runner state.
-
-        Raises:
-            WorkflowException: If in-flight messages or pending requests are present.
-        """
-        if await self._ctx.has_messages():
-            raise WorkflowException("Cannot capture checkpoint while in-flight messages are present.")
-
-        pending_requests = await self._ctx.get_pending_request_info_events()
-        if pending_requests:
-            raise WorkflowException("Cannot capture checkpoint while pending requests are present.")
-
-        await self._prepare_checkpoint_state()
-        return WorkflowCheckpoint(
-            workflow_name=self._workflow_name,
-            graph_signature_hash=self._graph_signature_hash,
-            previous_checkpoint_id=None,
-            messages={},
-            state=self._state.export_state(),
-            pending_request_info_events={},
-            iteration_count=0,
-            metadata=metadata or {},
-        )
-
     async def create_checkpoint_if_enabled(self) -> None:
         """Create a checkpoint if checkpointing is enabled and attach a label and metadata."""
         if not self._ctx.has_checkpointing():
@@ -332,32 +278,6 @@ class Runner:
                 self._previous_checkpoint_id,
             )
 
-    async def restore_from_checkpoint_object(self, checkpoint: WorkflowCheckpoint) -> None:
-        """Restore runner state from an in-memory checkpoint object.
-
-        Unlike :meth:`restore_from_checkpoint`, this does not load from storage or
-        validate the graph signature; it applies a checkpoint that the caller already
-        holds (for example, the workflow's captured initial checkpoint used by reset).
-
-        This clears any runtime checkpoint storage override and resets the context for a
-        fresh run, then restores shared state, executor snapshots, and runtime bookkeeping
-        from the checkpoint.
-
-        Args:
-            checkpoint: The checkpoint whose state should be restored.
-        """
-        self._ctx.clear_runtime_checkpoint_storage()
-        self._ctx.reset_for_new_run()
-
-        self._state.clear()
-        self._state.import_state(checkpoint.state)
-        await self._restore_executor_states()
-        self.reset_runtime_state(
-            iteration=checkpoint.iteration_count,
-            previous_checkpoint_id=checkpoint.previous_checkpoint_id,
-            resumed_from_checkpoint=False,
-        )
-
     async def restore_from_checkpoint(
         self,
         checkpoint_id: CheckpointID,

python/packages/core/agent_framework/_workflows/_runner_context.py

@@ -403,14 +403,12 @@ class InProcRunnerContext:
     def reset_for_new_run(self) -> None:
         """Reset the context for a new workflow run.
 
-        Clears messages, the pending event queue, the pending request_info
-        correlation map, and the streaming flag. Runtime checkpoint storage is
-        NOT cleared here as it's managed at the workflow level.
+        This clears messages, events, and resets streaming flag.
+        Runtime checkpoint storage is NOT cleared here as it's managed at the workflow level.
         """
         self._messages.clear()
         # Clear any pending events (best-effort) by recreating the queue
         self._event_queue = asyncio.Queue()
-        self._pending_request_info_events.clear()
         self._streaming = False  # Reset streaming flag
 
     async def apply_checkpoint(self, checkpoint: WorkflowCheckpoint) -> None:

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -20,7 +20,7 @@ from .._sessions import ContextProvider
 from .._types import ResponseStream
 from ..exceptions import WorkflowException
 from ..observability import OtelAttr, capture_exception, create_workflow_span
-from ._checkpoint import CheckpointStorage, WorkflowCheckpoint
+from ._checkpoint import CheckpointStorage
 from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
 from ._edge import (
     EdgeGroup,
@@ -371,10 +371,6 @@ class Workflow(DictConvertible):
         # so a subsequent ``run()`` is allowed.
         self._active_run: weakref.ref[ResponseStream[WorkflowEvent, WorkflowRunResult]] | None = None
 
-        # In-memory initial checkpoint captured from the just-built workflow state.
-        # This is internal-only and used by ``reset()``.
-        self._initial_checkpoint: WorkflowCheckpoint | None = None
-
     @property
     def status(self) -> WorkflowRunState:
         """Return the current run-level status of this workflow instance.
@@ -478,44 +474,6 @@ class Workflow(DictConvertible):
         """Get the list of executors in the workflow."""
         return list(self.executors.values())
 
-    async def _ensure_initial_checkpoint(self) -> None:
-        """Capture the in-memory initial checkpoint once for this workflow instance."""
-        if self._initial_checkpoint is not None:
-            return
-
-        self._initial_checkpoint = await self._runner.capture_checkpoint_object(
-            metadata={"kind": "initial_in_memory"},
-        )
-
-    async def reset(self) -> None:
-        """Reset the workflow instance to its captured initial checkpoint state.
-
-        The initial checkpoint is captured in memory once per workflow instance and
-        is not persisted to external checkpoint storage.
-
-        Raises:
-            WorkflowException: If called while a workflow run is active.
-        """
-        if self._is_run_active():
-            raise WorkflowException(
-                "Cannot reset workflow while a run is active. "
-                "Reset is only allowed between runs when the workflow is idle."
-            )
-
-        # Capture the baseline if it doesn't exist yet. This is idempotent: on a
-        # normal reset after one or more runs it's a no-op (the snapshot was taken
-        # before the first run); when reset is the first operation it captures the
-        # pristine just-built state so the workflow stays runnable.
-        await self._ensure_initial_checkpoint()
-        if self._initial_checkpoint is None:
-            raise WorkflowException("Workflow initial checkpoint is unavailable.")
-
-        # Restore runner state, executor snapshots, and runtime bookkeeping from the
-        # in-memory initial checkpoint.
-        await self._runner.restore_from_checkpoint_object(self._initial_checkpoint)
-
-        self._status = WorkflowRunState.IDLE
-
     async def _run_workflow_with_tracing(
         self,
         initial_executor_fn: Callable[[], Awaitable[None]] | None = None,
@@ -798,6 +756,12 @@ class Workflow(DictConvertible):
                 "Workflow is already running; concurrent runs are not allowed on the same instance."
             )
 
+        # No run is active, so any runtime checkpoint storage override still set on the
+        # context is stale - left over from a prior run whose stream was dropped before
+        # its async-generator finalizer ran. Clear it so this run starts clean and does
+        # not silently inherit the prior run's runtime checkpoint storage.
+        self._runner.context.clear_runtime_checkpoint_storage()
+
         response_stream = ResponseStream[WorkflowEvent, WorkflowRunResult](
             self._run_core(
                 message=message,
@@ -832,10 +796,6 @@ class Workflow(DictConvertible):
         Yields:
             WorkflowEvent: The events generated during the workflow execution.
         """
-        # Enable runtime checkpointing if storage provided
-        if checkpoint_storage is not None:
-            self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
-
         # Capture the weakref instance ``run()`` installed for *this* run. We
         # compare by object identity in the finally so a stale finalizer (e.g.
         # the caller dropped this stream after partial iteration, then started
@@ -846,6 +806,10 @@ class Workflow(DictConvertible):
         # here it already points at our own ``ResponseStream``.
         my_active_run = self._active_run
 
+        # Enable runtime checkpointing if storage provided.
+        if checkpoint_storage is not None:
+            self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
+
         try:
             # Async validation: a fresh-message run is only allowed when the
             # runner context has fully drained from any prior run. If it still
@@ -867,8 +831,6 @@ class Workflow(DictConvertible):
                     "checkpointing; there is no in-process recovery path."
                 )
 
-            await self._ensure_initial_checkpoint()
-
             initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
 
             async for event in self._run_workflow_with_tracing(
@@ -889,16 +851,24 @@ class Workflow(DictConvertible):
                     continue
                 yield event
         finally:
-            # Clear the active-run weakref so a subsequent ``run()`` is allowed,
-            # but only if the slot still holds *our* weakref. If the caller
-            # dropped this stream after partial iteration and a new ``run()``
-            # already installed its own weakref before our async-gen finalizer
-            # ran, ``self._active_run`` now points at the successor; clearing
-            # it would silently break the successor's concurrency guard.
-            if self._active_run is my_active_run:
+            # Whether this run is still the active one (no successor ``run()`` has
+            # installed a new weakref since we started). Captured once because the
+            # active-run clear below mutates ``self._active_run``. Used to scope both
+            # the run-lock release and the runtime-storage clear so a dropped run's
+            # deferred finalizer cannot clobber a successor run's state.
+            owns_run = self._active_run is my_active_run
+            if owns_run:
+                # Clear the active-run weakref so a subsequent ``run()`` is allowed.
+                # If the caller dropped this stream after partial iteration and a new
+                # ``run()`` already installed its own weakref before our async-gen
+                # finalizer ran, ``self._active_run`` points at the successor and we
+                # leave it untouched to preserve the successor's concurrency guard.
                 self._active_run = None
-            if checkpoint_storage is not None:
-                self._runner.context.clear_runtime_checkpoint_storage()
+                # Same ownership scoping applies to the runtime checkpoint storage:
+                # only clear it when this run still owns it, so a dropped run's
+                # deferred finalizer can't clear a successor's storage.
+                if checkpoint_storage is not None:
+                    self._runner.context.clear_runtime_checkpoint_storage()
 
     @staticmethod
     def _finalize_events(

python/packages/core/agent_framework/_workflows/_workflow_executor.py

@@ -517,10 +517,6 @@ class WorkflowExecutor(Executor):
         self._execution_contexts = execution_contexts
         self._request_to_execution = request_to_execution
 
-        # Reset the sub workflow to its initial state. This must be done before pumping
-        # the request info events back into the sub workflow.
-        await self.workflow.reset()
-
         # Add the `request_info_event`s back to the sub workflow.
         # This is only a temporary solution to rehydrate the sub workflow with the requests.
         # The proper way would be to rehydrate the workflow from a checkpoint on a Workflow

python/packages/core/agent_framework/observability.py

@@ -2211,6 +2211,181 @@ def _get_instructions_from_options(options: Any) -> str | list[str] | None:
     return None
 
 
+# region OTel tool definitions
+
+# Per-item in-memory cache of computed OTel tool definitions, keyed by the tool
+# object's identity. Tool objects (e.g. ``FunctionTool``, ``MCPTool``) are often
+# reused across runs, so caching their converted definitions avoids repeating the
+# isinstance checks, schema generation, and dict construction on every invocation.
+# A ``WeakKeyDictionary`` lets entries be garbage collected with their tools.
+# Unhashable / non-weak-referenceable specs (e.g. plain dicts) bypass the cache.
+_TOOL_OTEL_DEFINITION_CACHE: weakref.WeakKeyDictionary[Any, dict[str, Any] | None] = weakref.WeakKeyDictionary()
+# Sentinel distinguishing "not cached" from a cached ``None`` (unparseable tool).
+_CACHE_MISS: Final = object()
+
+
+def _tools_to_dict(
+    tools: Any,
+) -> list[dict[str, Any]] | None:
+    """Convert tools into OpenTelemetry GenAI tool definitions.
+
+    The output conforms to the OTel GenAI tool-definitions schema, where each
+    entry is either a ``FunctionToolDefinition`` (``type="function"`` with
+    ``name`` and optional ``description``/``parameters``) or a
+    ``GenericToolDefinition`` (any ``type`` plus a ``name``). See
+    https://github.com/open-telemetry/semantic-conventions-genai/blob/main/docs/gen-ai/gen-ai-tool-definitions.json.
+
+    Args:
+        tools: The tools to parse. Can be a single tool or a sequence of tools.
+
+    Returns:
+        A list of OTel-conformant tool-definition dicts, or ``None`` when
+        ``tools`` is empty or no tool can be represented.
+    """
+    from ._tools import normalize_tools
+
+    normalized_tools = normalize_tools(tools)
+    if not normalized_tools:
+        return None
+    results: list[dict[str, Any]] = []
+    for tool_item in normalized_tools:
+        otel_def = _tool_to_otel_definition(tool_item)
+        if otel_def is not None:
+            results.append(otel_def)
+    return results or None
+
+
+def _tool_to_otel_definition(tool_item: Any) -> dict[str, Any] | None:
+    """Convert a single tool spec into an OTel GenAI tool-definition dict.
+
+    Results are cached per tool object (keyed by identity) so repeated runs that
+    reuse the same tool instances skip the conversion work. Specs that cannot be
+    weakly referenced (e.g. plain dicts) are converted without caching.
+
+    Returns ``None`` and emits a warning when the input cannot be represented
+    as either a ``FunctionToolDefinition`` or a ``GenericToolDefinition``.
+    """
+    try:
+        cached = _TOOL_OTEL_DEFINITION_CACHE.get(tool_item, _CACHE_MISS)
+    except TypeError:
+        # Unhashable spec (e.g. a plain dict); convert without caching.
+        return _build_tool_otel_definition(tool_item)
+    if cached is not _CACHE_MISS:
+        return cast("dict[str, Any] | None", cached)
+
+    definition = _build_tool_otel_definition(tool_item)
+    with contextlib.suppress(TypeError):
+        # Object may not support weak references; skip caching when that is the case.
+        _TOOL_OTEL_DEFINITION_CACHE[tool_item] = definition
+    return definition
+
+
+def _build_tool_otel_definition(tool_item: Any) -> dict[str, Any] | None:
+    """Convert a single tool spec into an OTel GenAI tool-definition dict (uncached)."""
+    from pydantic import BaseModel
+
+    from ._mcp import MCPTool
+    from ._serialization import SerializationMixin
+    from ._tools import FunctionTool
+
+    if isinstance(tool_item, FunctionTool):
+        definition: dict[str, Any] = {"type": "function", "name": tool_item.name}
+        if tool_item.description:
+            definition["description"] = tool_item.description
+        parameters = tool_item.parameters()
+        if parameters:
+            definition["parameters"] = parameters
+        return definition
+
+    if isinstance(tool_item, MCPTool):
+        definition = {"type": "mcp", "name": tool_item.name}
+        if tool_item.description:
+            definition["description"] = tool_item.description
+        return definition
+
+    raw: Mapping[str, Any] | None = None
+    if isinstance(tool_item, BaseModel):
+        raw = tool_item.model_dump(exclude_none=True)
+    elif isinstance(tool_item, SerializationMixin):
+        raw = tool_item.to_dict()
+    elif isinstance(tool_item, Mapping):
+        raw = cast("Mapping[str, Any]", tool_item)
+
+    if raw is None:
+        logger.warning(
+            "Can't parse tool to OpenTelemetry tool definition: %s",
+            type(tool_item).__name__,  # type: ignore[reportUnknownArgumentType]
+        )
+        return None
+    return _otel_definition_from_mapping(raw)
+
+
+def _otel_definition_from_mapping(raw: Mapping[str, Any]) -> dict[str, Any] | None:
+    """Reshape a tool spec mapping into an OTel GenAI tool-definition dict.
+
+    Handles the nested OpenAI Chat Completions function shape
+    (``{"type": "function", "function": {...}}``) by flattening it into the
+    OTel shape.
+    """
+    # OpenAI Chat Completions nests the function spec one level deeper; flatten it.
+    nested_function = raw.get("function") if raw.get("type") == "function" else None
+    if isinstance(nested_function, Mapping):
+        nested = cast("Mapping[str, Any]", nested_function)
+        name = nested.get("name")
+        if not isinstance(name, str) or not name:
+            logger.warning("Can't parse tool to OpenTelemetry tool definition: missing 'name'.")
+            return None
+        definition: dict[str, Any] = {"type": "function", "name": name}
+        description = nested.get("description")
+        if description:
+            definition["description"] = description
+        parameters = nested.get("parameters")
+        if parameters:
+            definition["parameters"] = parameters
+        # Forward extra properties from both layers, preferring the inner spec.
+        for source in (nested, raw):
+            for key, value in source.items():
+                if key in {"type", "function", "name", "description", "parameters"}:
+                    continue
+                definition.setdefault(key, value)
+        return definition
+
+    type_value = raw.get("type")
+    if not isinstance(type_value, str) or not type_value:
+        logger.warning("Can't parse tool to OpenTelemetry tool definition: missing 'type'.")
+        return None
+
+    name_value = raw.get("name")
+    if not isinstance(name_value, str) or not name_value:
+        # Hosted tools sometimes omit ``name`` (e.g. ``{"type": "code_interpreter"}``);
+        # fall back to the type so the OTel definition stays valid.
+        name_value = type_value
+
+    if type_value == "function":
+        definition = {"type": "function", "name": name_value}
+        description = raw.get("description")
+        if description:
+            definition["description"] = description
+        parameters = raw.get("parameters")
+        if parameters:
+            definition["parameters"] = parameters
+        for key, value in raw.items():
+            if key in {"type", "name", "description", "parameters"}:
+                continue
+            definition.setdefault(key, value)
+        return definition
+
+    definition = {"type": type_value, "name": name_value}
+    for key, value in raw.items():
+        if key in {"type", "name"}:
+            continue
+        definition[key] = value
+    return definition
+
+
+# endregion
+
+
 # Mapping configuration for extracting span attributes
 # Each entry: source_keys -> (otel_attribute_key, transform_func, check_options_first, default_value)
 # - source_keys: single key or list of keys to check (first non-None value wins)
@@ -2246,11 +2421,7 @@ OTEL_ATTR_MAP: dict[str | tuple[str, ...], tuple[str, Callable[[Any], Any] | Non
     # Tools with validation - returns None if no valid tools
     "tools": (
         OtelAttr.TOOL_DEFINITIONS,
-        lambda tools: (
-            json.dumps(tools_dict, ensure_ascii=False)
-            if (tools_dict := __import__("agent_framework._tools", fromlist=["_tools_to_dict"])._tools_to_dict(tools))
-            else None
-        ),
+        lambda tools: json.dumps(tools_dict, ensure_ascii=False) if (tools_dict := _tools_to_dict(tools)) else None,
         True,
         None,
     ),

python/packages/core/tests/core/test_mcp.py

@@ -122,6 +122,113 @@ async def test_load_tools_with_tool_name_prefix_preserves_matching_configuration
     assert tool.functions[0].approval_mode == "always_require"
 
 
+async def test_allowed_tools_does_not_authorize_normalized_remote_name_collision() -> None:
+    """A normalized/local allowlist match must not authorize a different raw remote tool."""
+    tool = MCPTool(name="test_server", allowed_tools=["delete-file"])  # type: ignore[abstract]
+
+    mock_session = AsyncMock()
+    tool.session = mock_session
+    tool.load_tools_flag = True
+
+    page = Mock()
+    page.tools = [
+        types.Tool(
+            name="delete/file",
+            description="Delete a file",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+    ]
+    page.nextCursor = None
+    mock_session.list_tools = AsyncMock(return_value=page)
+
+    await tool.load_tools()
+
+    assert [function.name for function in tool._functions] == ["delete-file"]
+    assert tool.functions == []
+
+
+async def test_load_tools_rejects_colliding_normalized_tool_names() -> None:
+    """A remote MCP server must not choose which raw tool backs a colliding local name."""
+    tool = MCPTool(name="test_server", allowed_tools=["delete-file"])  # type: ignore[abstract]
+
+    mock_session = AsyncMock()
+    tool.session = mock_session
+    tool.load_tools_flag = True
+
+    page = Mock()
+    page.tools = [
+        types.Tool(
+            name="delete/file",
+            description="Unauthorized tool",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+        types.Tool(
+            name="delete-file",
+            description="Authorized tool",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+    ]
+    page.nextCursor = None
+    mock_session.list_tools = AsyncMock(return_value=page)
+
+    with pytest.raises(ToolExecutionException, match="map to the same local function name"):
+        await tool.load_tools()
+
+
+async def test_allowed_tools_exact_raw_name_allows_normalized_function_name() -> None:
+    """An exact raw remote allowlist entry still exposes that raw tool, regardless of local normalization."""
+    tool = MCPTool(name="test_server", allowed_tools=["delete/file"])  # type: ignore[abstract]
+
+    mock_session = AsyncMock()
+    tool.session = mock_session
+    tool.load_tools_flag = True
+
+    page = Mock()
+    page.tools = [
+        types.Tool(
+            name="delete/file",
+            description="Delete a file",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+    ]
+    page.nextCursor = None
+    mock_session.list_tools = AsyncMock(return_value=page)
+
+    await tool.load_tools()
+
+    assert [function.name for function in tool.functions] == ["delete-file"]
+    assert tool.functions[0].additional_properties is not None
+    assert tool.functions[0].additional_properties["_mcp_remote_name"] == "delete/file"
+
+
+async def test_approval_mode_does_not_match_normalized_colliding_name() -> None:
+    """Approval rules should not apply to a different raw remote tool through normalization."""
+    tool = MCPTool(  # type: ignore[abstract]
+        name="test_server",
+        approval_mode={"always_require_approval": ["delete-file"]},
+    )
+
+    mock_session = AsyncMock()
+    tool.session = mock_session
+    tool.load_tools_flag = True
+
+    page = Mock()
+    page.tools = [
+        types.Tool(
+            name="delete/file",
+            description="Delete a file",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+    ]
+    page.nextCursor = None
+    mock_session.list_tools = AsyncMock(return_value=page)
+
+    await tool.load_tools()
+
+    assert tool._functions[0].name == "delete-file"
+    assert tool._functions[0].approval_mode == "never_require"
+
+
 async def test_load_prompts_with_tool_name_prefix() -> None:
     """Prefixed MCP prompt names should be exposed with the configured prefix."""
     tool = MCPTool(name="docs", tool_name_prefix="docs")  # type: ignore[abstract]
@@ -3339,6 +3446,7 @@ async def test_load_tools_adds_properties_to_zero_arg_tool_schema():
     none_schema_tool.name = "none_schema_tool"
     none_schema_tool.description = "A tool with None inputSchema"
     none_schema_tool.inputSchema = None
+    none_schema_tool.meta = None
     page.tools.append(none_schema_tool)
     page.nextCursor = None
 
@@ -4777,7 +4885,7 @@ async def test_mcp_tool_call_tool_forwards_tool_list_meta():
 
 
 async def test_mcp_tool_call_tool_user_meta_merges_with_tool_list_meta():
-    """User-provided _meta should be sent as MCP request metadata, not tool arguments."""
+    """Tools/list _meta should win over caller-provided _meta on conflicts."""
     from opentelemetry import trace
 
     tool_meta = {"from_tool": "tool-value", "shared": "tool-value"}
@@ -4817,11 +4925,153 @@ async def test_mcp_tool_call_tool_user_meta_merges_with_tool_list_meta():
         assert call_kwargs["meta"] == {
             "from_tool": "tool-value",
             "from_user": "user-value",
-            "shared": "user-value",
+            "shared": "tool-value",
         }
         assert user_meta == {"from_user": "user-value", "shared": "user-value"}
 
 
+async def test_mcp_tool_function_invocation_strips_model_supplied_meta() -> None:
+    """Model-supplied _meta should not become MCP request metadata."""
+    from opentelemetry import trace
+
+    class TestServer(MCPTool):
+        async def connect(self) -> None:  # type: ignore[override]  # pyrefly: ignore[bad-override]  # ty: ignore[invalid-method-override]
+            self.session = Mock(spec=ClientSession)
+            self.session.list_tools = AsyncMock(
+                return_value=types.ListToolsResult(
+                    tools=[
+                        types.Tool(
+                            name="test_tool",
+                            description="Test tool",
+                            inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
+                        )
+                    ]
+                )
+            )
+            self.session.call_tool = AsyncMock(
+                return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
+            )
+
+        def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
+            return None  # type: ignore[return-value]  # pyrefly: ignore[bad-return]  # ty: ignore[invalid-return-type]
+
+    server = TestServer(name="test_server")
+    async with server:
+        await server.load_tools()
+
+        with (
+            trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
+            patch("agent_framework._mcp.propagate.inject", side_effect=lambda carrier: None),
+        ):
+            await server.functions[0].invoke(
+                arguments={"param": "test_value", "_meta": {"attacker.example/route": "evil"}}
+            )
+
+        call_kwargs = server.session.call_tool.call_args.kwargs  # type: ignore[union-attr]  # ty: ignore[unresolved-attribute]
+        assert call_kwargs["arguments"] == {"param": "test_value"}
+        assert call_kwargs["meta"] is None
+
+
+async def test_mcp_tool_function_invocation_preserves_trusted_meta_over_model_meta() -> None:
+    """Trusted function-invocation _meta should be restored after model arguments are merged."""
+    from opentelemetry import trace
+
+    trusted_meta = {"trusted.example/route": "trusted"}
+
+    class TestServer(MCPTool):
+        async def connect(self) -> None:  # type: ignore[override]  # pyrefly: ignore[bad-override]  # ty: ignore[invalid-method-override]
+            self.session = Mock(spec=ClientSession)
+            self.session.list_tools = AsyncMock(
+                return_value=types.ListToolsResult(
+                    tools=[
+                        types.Tool(
+                            name="test_tool",
+                            description="Test tool",
+                            inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
+                        )
+                    ]
+                )
+            )
+            self.session.call_tool = AsyncMock(
+                return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
+            )
+
+        def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
+            return None  # type: ignore[return-value]  # pyrefly: ignore[bad-return]  # ty: ignore[invalid-return-type]
+
+    server = TestServer(name="test_server")
+    async with server:
+        await server.load_tools()
+
+        context = FunctionInvocationContext(
+            function=server.functions[0],
+            arguments={},
+            kwargs={"_meta": trusted_meta},
+        )
+        with (
+            trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
+            patch("agent_framework._mcp.propagate.inject", side_effect=lambda carrier: None),
+        ):
+            await server.functions[0].invoke(
+                arguments={"param": "test_value", "_meta": {"attacker.example/route": "evil"}},
+                context=context,
+            )
+
+        call_kwargs = server.session.call_tool.call_args.kwargs  # type: ignore[union-attr]  # ty: ignore[unresolved-attribute]
+        assert call_kwargs["arguments"] == {"param": "test_value"}
+        assert call_kwargs["meta"] == trusted_meta
+
+
+async def test_mcp_tool_call_tool_otel_meta_overrides_user_meta_but_not_tool_list_meta() -> None:
+    """OpenTelemetry should override caller metadata while tools/list metadata remains most trusted."""
+    from opentelemetry import trace
+
+    tool_meta = {"traceparent": "tool-traceparent", "from_tool": "tool-value"}
+    user_meta = {"traceparent": "user-traceparent", "from_user": "user-value"}
+
+    class TestServer(MCPTool):
+        async def connect(self) -> None:  # type: ignore[override]  # pyrefly: ignore[bad-override]  # ty: ignore[invalid-method-override]
+            self.session = Mock(spec=ClientSession)
+            self.session.list_tools = AsyncMock(
+                return_value=types.ListToolsResult(
+                    tools=[
+                        types.Tool(
+                            name="test_tool",
+                            description="Test tool",
+                            inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
+                            _meta=tool_meta,
+                        )
+                    ]
+                )
+            )
+            self.session.call_tool = AsyncMock(
+                return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
+            )
+
+        def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
+            return None  # type: ignore[return-value]  # pyrefly: ignore[bad-return]  # ty: ignore[invalid-return-type]
+
+    server = TestServer(name="test_server")
+    async with server:
+        await server.load_tools()
+
+        with (
+            trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
+            patch(
+                "agent_framework._mcp.propagate.inject",
+                side_effect=lambda carrier: carrier.update({"traceparent": "otel-traceparent"}),
+            ),
+        ):
+            await server.call_tool("test_tool", param="test_value", _meta=user_meta)
+
+        call_kwargs = server.session.call_tool.call_args.kwargs  # type: ignore[union-attr]  # ty: ignore[unresolved-attribute]
+        assert call_kwargs["meta"] == {
+            "traceparent": "tool-traceparent",
+            "from_tool": "tool-value",
+            "from_user": "user-value",
+        }
+
+
 async def test_mcp_streamable_http_tool_hook_not_duplicated_on_repeated_get_mcp_client():
     """Test that calling get_mcp_client multiple times does not accumulate duplicate hooks."""
     tool = MCPStreamableHTTPTool(
@@ -6475,6 +6725,30 @@ def test_prepare_call_kwargs_extracts_meta() -> None:
     assert meta.get("trace") == "abc"
 
 
+@pytest.mark.parametrize(
+    "key",
+    [
+        "",
+        "_leading-underscore",
+        "trailing-underscore_",
+        "abc/",
+        "1bad.example/name",
+        "bad..example/name",
+        "bad.example/_name",
+        "bad.example/name_",
+    ],
+)
+def test_prepare_call_kwargs_rejects_invalid_meta_key_names(key: str) -> None:
+    server = MCPTool(name="test_server")  # type: ignore[abstract]
+    server._tool_param_names_by_name = {"test_tool": {"param"}}
+
+    with pytest.raises(ToolExecutionException, match="Invalid MCP _meta key name"):
+        server._prepare_call_kwargs(
+            "test_tool",
+            {"param": "v", "_meta": {key: "value"}},
+        )
+
+
 async def test_call_tool_forwards_only_declared_arguments() -> None:
     """End-to-end: framework runtime kwargs are stripped before reaching the server."""
 

python/packages/core/tests/core/test_observability.py

@@ -3132,6 +3132,223 @@ def test_get_span_attributes_with_agent_info():
     assert attrs[OtelAttr.AGENT_DESCRIPTION] == "A test agent"
 
 
+def test_get_span_attributes_emits_otel_tool_definitions() -> None:
+    """``tools`` are serialized to OTel GenAI tool definitions on the span."""
+    import json as _json
+
+    from agent_framework import tool
+    from agent_framework.observability import OtelAttr, _get_span_attributes
+
+    @tool(name="echo", description="Echo input")
+    def echo(value: str) -> str:
+        return value
+
+    attrs = _get_span_attributes(
+        operation_name="chat",
+        provider_name="openai",
+        model="gpt-4",
+        service_url="https://api.openai.com",
+        tools=[
+            echo,
+            {
+                "type": "function",
+                "function": {
+                    "name": "lookup",
+                    "description": "Lookup by id",
+                    "parameters": {"type": "object", "properties": {"id": {"type": "string"}}},
+                },
+            },
+            {"type": "web_search", "name": "web_search"},
+        ],
+    )
+
+    assert OtelAttr.TOOL_DEFINITIONS in attrs
+    definitions = _json.loads(attrs[OtelAttr.TOOL_DEFINITIONS])
+    assert definitions == [
+        {
+            "type": "function",
+            "name": "echo",
+            "description": "Echo input",
+            "parameters": echo.parameters(),
+        },
+        {
+            "type": "function",
+            "name": "lookup",
+            "description": "Lookup by id",
+            "parameters": {"type": "object", "properties": {"id": {"type": "string"}}},
+        },
+        {"type": "web_search", "name": "web_search"},
+    ]
+
+
+def test_get_span_attributes_omits_tool_definitions_when_unparseable() -> None:
+    """When no tool can be converted, the tool definitions attribute is omitted."""
+    from agent_framework.observability import OtelAttr, _get_span_attributes
+
+    attrs = _get_span_attributes(
+        operation_name="chat",
+        provider_name="openai",
+        model="gpt-4",
+        service_url="https://api.openai.com",
+        tools=[{"kind": "not_an_otel_tool"}],
+    )
+
+    assert OtelAttr.TOOL_DEFINITIONS not in attrs
+
+
+def test_tools_to_dict_supports_pydantic_tool_models() -> None:
+    """Pydantic-based tool specs are reshaped into the OTel GenAI tool-definition shape."""
+    from pydantic import BaseModel
+
+    from agent_framework.observability import _tools_to_dict
+
+    class ProviderTool(BaseModel):
+        type: str
+        name: str
+        enabled: bool = True
+        note: str | None = None
+
+    result = _tools_to_dict([ProviderTool(type="web_search", name="web_search")])
+
+    assert result == [{"type": "web_search", "name": "web_search", "enabled": True}]
+
+
+def test_tools_to_dict_returns_none_for_empty_input() -> None:
+    """``_tools_to_dict`` returns None when no tools are supplied."""
+    from agent_framework.observability import _tools_to_dict
+
+    assert _tools_to_dict(None) is None
+    assert _tools_to_dict([]) is None
+
+
+def test_tools_to_dict_function_tool_uses_otel_function_definition() -> None:
+    """``FunctionTool`` instances are emitted as flat OTel FunctionToolDefinition dicts."""
+    from agent_framework import tool
+    from agent_framework.observability import _tools_to_dict
+
+    @tool(name="add", description="Add two numbers")
+    def add(x: int, y: int) -> int:
+        return x + y
+
+    result = _tools_to_dict([add])
+
+    assert result is not None
+    assert len(result) == 1
+    definition = result[0]
+    assert definition["type"] == "function"
+    assert definition["name"] == "add"
+    assert definition["description"] == "Add two numbers"
+    assert definition["parameters"]["type"] == "object"
+    assert set(definition["parameters"]["required"]) == {"x", "y"}
+    # The legacy OpenAI Chat Completions ``function`` wrapper is not part of the OTel shape.
+    assert "function" not in definition
+
+
+def test_tools_to_dict_flattens_openai_chat_completions_function_spec() -> None:
+    """OpenAI Chat Completions nested ``function`` spec is flattened to the OTel shape."""
+    from agent_framework.observability import _tools_to_dict
+
+    openai_spec = {
+        "type": "function",
+        "function": {
+            "name": "lookup_user",
+            "description": "Look up a user by id",
+            "parameters": {
+                "type": "object",
+                "properties": {"user_id": {"type": "string"}},
+                "required": ["user_id"],
+            },
+            "strict": True,
+        },
+    }
+
+    result = _tools_to_dict([openai_spec])
+
+    assert result == [
+        {
+            "type": "function",
+            "name": "lookup_user",
+            "description": "Look up a user by id",
+            "parameters": {
+                "type": "object",
+                "properties": {"user_id": {"type": "string"}},
+                "required": ["user_id"],
+            },
+            "strict": True,
+        }
+    ]
+
+
+def test_tools_to_dict_passes_through_hosted_tool_dicts() -> None:
+    """Hosted-tool dicts pass through with the OTel required keys preserved."""
+    from agent_framework.observability import _tools_to_dict
+
+    result = _tools_to_dict([{"type": "web_search", "name": "web_search", "max_results": 5}])
+
+    assert result == [{"type": "web_search", "name": "web_search", "max_results": 5}]
+
+
+def test_tools_to_dict_falls_back_to_type_when_name_missing() -> None:
+    """Hosted-tool dicts without ``name`` fall back to the ``type`` value."""
+    from agent_framework.observability import _tools_to_dict
+
+    result = _tools_to_dict([{"type": "code_interpreter"}])
+
+    assert result == [{"type": "code_interpreter", "name": "code_interpreter"}]
+
+
+def test_tools_to_dict_warns_when_type_missing(caplog: pytest.LogCaptureFixture) -> None:
+    """Tools without an extractable ``type`` are skipped with a warning."""
+    from agent_framework.observability import _tools_to_dict
+
+    with caplog.at_level("WARNING", logger="agent_framework"):
+        result = _tools_to_dict([{"kind": "not_an_otel_tool"}])
+
+    assert result is None
+    assert any("missing 'type'" in rec.message for rec in caplog.records)
+
+
+def test_tools_to_dict_warns_for_unknown_tool_object(caplog: pytest.LogCaptureFixture) -> None:
+    """Tools that are neither callable, mapping, BaseModel, nor known type are skipped."""
+    from agent_framework.observability import _tools_to_dict
+
+    class _Opaque:
+        pass
+
+    with caplog.at_level("WARNING", logger="agent_framework"):
+        result = _tools_to_dict([_Opaque()])
+
+    assert result is None
+    assert any("OpenTelemetry tool definition" in rec.message for rec in caplog.records)
+
+
+def test_tool_to_otel_definition_caches_per_tool_object() -> None:
+    """Converting the same tool object twice reuses the cached OTel definition."""
+    from agent_framework import tool
+    from agent_framework.observability import _build_tool_otel_definition, _tool_to_otel_definition
+
+    @tool(name="add", description="Add two numbers")
+    def add(x: int, y: int) -> int:
+        return x + y
+
+    first = _tool_to_otel_definition(add)
+    second = _tool_to_otel_definition(add)
+
+    # The cached result is returned as the same object on subsequent conversions.
+    assert first is second
+    # A fresh (uncached) build produces an equal but distinct object.
+    assert _build_tool_otel_definition(add) == first
+
+
+def test_tool_to_otel_definition_skips_cache_for_unhashable_specs() -> None:
+    """Plain-dict tool specs are converted without raising despite being uncacheable."""
+    from agent_framework.observability import _tool_to_otel_definition
+
+    spec = {"type": "web_search", "name": "web_search"}
+
+    assert _tool_to_otel_definition(spec) == {"type": "web_search", "name": "web_search"}
+
+
 # region Test _capture_response
 
 

python/packages/core/tests/core/test_tools.py

@@ -19,26 +19,12 @@ from agent_framework._middleware import FunctionInvocationContext
 from agent_framework._tools import (
     _parse_annotation,
     _parse_inputs,
-    _tools_to_dict,
 )
 from agent_framework.observability import OtelAttr
 
 # region FunctionTool and tool decorator tests
 
 
-def test_tools_to_dict_supports_pydantic_tool_models() -> None:
-    """Pydantic-based tool specs are serialized without logging parse warnings."""
-
-    class ProviderTool(BaseModel):
-        kind: str
-        enabled: bool = True
-        note: str | None = None
-
-    result = _tools_to_dict([ProviderTool(kind="google_search")])
-
-    assert result == [{"kind": "google_search", "enabled": True}]
-
-
 def test_tool_decorator():
     """Test the tool decorator."""
 

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

@@ -3860,6 +3860,61 @@ class TestResponseStreamMapAndWithFinalizer:
         final = await outer.get_final_response()
         assert final.text == "mapped_update_0mapped_update_1"
 
+    async def test_flat_map_expands_updates(self) -> None:
+        """flat_map() can transform one update into many updates."""
+        inner = ResponseStream(_generate_updates(2), finalizer=_combine_updates)
+
+        def expand(update: ChatResponseUpdate) -> list[ChatResponseUpdate]:
+            return [
+                ChatResponseUpdate(contents=[Content.from_text(update.text)], role=cast(Any, update.role)),
+                ChatResponseUpdate(contents=[Content.from_text(f"{update.text}_extra")], role=cast(Any, update.role)),
+            ]
+
+        outer = inner.flat_map(expand, _combine_updates)
+
+        collected: list[str] = []
+        async for update in outer:
+            collected.append(update.text or "")
+
+        assert collected == ["update_0", "update_0_extra", "update_1", "update_1_extra"]
+
+        final = await outer.get_final_response()
+        assert final.text == "update_0update_0_extraupdate_1update_1_extra"
+
+    async def test_flat_map_skips_empty_mappings(self) -> None:
+        """flat_map() supports zero-output transforms."""
+        inner = ResponseStream(_generate_updates(3), finalizer=_combine_updates)
+
+        def keep_odd(update: ChatResponseUpdate) -> list[ChatResponseUpdate]:
+            return [update] if update.text == "update_1" else []
+
+        outer = inner.flat_map(keep_odd, _combine_updates)
+
+        collected = [update.text async for update in outer]
+        assert collected == ["update_1"]
+
+        final = await outer.get_final_response()
+        assert final.text == "update_1"
+
+    async def test_flat_map_calls_inner_result_hooks(self) -> None:
+        """flat_map() preserves inner result hooks."""
+        inner_result_hook_called = {"value": False}
+
+        def inner_result_hook(response: ChatResponse) -> ChatResponse:
+            inner_result_hook_called["value"] = True
+            return response
+
+        inner = ResponseStream(
+            _generate_updates(2),
+            finalizer=_combine_updates,
+            result_hooks=[inner_result_hook],  # ty: ignore[invalid-argument-type]
+        )
+        outer = inner.flat_map(lambda u: [u], _combine_updates)
+
+        await outer.get_final_response()
+
+        assert inner_result_hook_called["value"] is True
+
     async def test_outer_transform_hooks_independent(self) -> None:
         """Outer stream has its own independent transform hooks."""
         inner_hook_calls = {"value": 0}

python/packages/core/tests/workflow/test_checkpoint.py

@@ -355,7 +355,7 @@ async def test_workflow_checkpoint_ancestry_preserved_after_resume():
 
     class FinishExecutor(Executor):
         @handler
-        async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
+        async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:  # type: ignore[valid-type]
             await ctx.yield_output(message + "-done")
 
     storage = InMemoryCheckpointStorage()

python/packages/core/tests/workflow/test_checkpoint_validation.py

@@ -23,7 +23,7 @@ class StartExecutor(Executor):
 
 class FinishExecutor(Executor):
     @handler
-    async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:  # type: ignore[valid-type]
+    async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:  # zuban: ignore
         await ctx.yield_output(message)
 
 
@@ -95,7 +95,7 @@ class SubStartExecutor(Executor):
 
 class SubFinishExecutor(Executor):
     @handler
-    async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:  # type: ignore[valid-type]
+    async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None:  # zuban: ignore
         await ctx.yield_output(message)
 
 

python/packages/core/tests/workflow/test_runner.py

@@ -1058,6 +1058,172 @@ async def test_runner_drains_events_on_iteration_exception():
     assert len(output_events) >= 1
 
 
+async def test_runner_resumed_flag_reset_after_failed_resumed_run():
+    """A failed *resumed* run must not leak the resume flag into the next run.
+
+    The resume flag suppresses the initial "superstep 0" (entry) checkpoint when resuming from an
+    iteration-0 checkpoint (which already exists and must not be recreated). It used to be cleared
+    only on the success path, so an executor failure during a resumed run left it ``True`` and the
+    next fresh run wrongly skipped its entry checkpoint. The flag is now cleared in a ``finally`` so
+    this holds even when convergence raises.
+
+    This also verifies checkpoint creation on the re-run: the resumed (failed) run creates no entry
+    checkpoint, while the subsequent fresh run does.
+    """
+    storage = InMemoryCheckpointStorage()
+    ctx = CheckpointingContext(storage)
+    executor_a = PassthroughExecutor(id="executor_a")
+    executor_b = ExecutorThatFailsWithEvents(id="executor_b", runner_ctx=ctx, fail_on_iteration=1)
+
+    edges = [SingleEdgeGroup(executor_a.id, executor_b.id)]
+    executors: dict[str, Executor] = {executor_a.id: executor_a, executor_b.id: executor_b}
+    state = State()
+
+    runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Simulate a resumed run; this marks the runner as resumed so the next run skips
+    # the superstep-0 checkpoint.
+    resumed_checkpoint = WorkflowCheckpoint(
+        checkpoint_id="resumed-cp",
+        workflow_name="test_name",
+        graph_signature_hash="test_hash",
+        iteration_count=0,
+    )
+    runner._mark_resumed(resumed_checkpoint)  # pyright: ignore[reportPrivateUsage]
+    assert runner._resumed_from_checkpoint is True  # pyright: ignore[reportPrivateUsage]
+
+    # Run the resumed turn; executor_b fails mid-iteration before any superstep
+    # checkpoint is created.
+    await executor_a.execute(MockMessage(data=0), ["START"], state, ctx)
+    with pytest.raises(RuntimeError, match="Executor failed with pending events"):
+        async for _ in runner.run_until_convergence():
+            pass
+
+    # The fix: the resume flag is cleared even though the run raised.
+    assert runner._resumed_from_checkpoint is False  # pyright: ignore[reportPrivateUsage]
+    # The resumed (failed) run created no superstep-0 checkpoint (it was skipped).
+    assert await storage.list_checkpoints(workflow_name="test_name") == []
+
+    # Re-run as a fresh turn: with the flag correctly reset, the runner now creates
+    # the initial superstep-0 checkpoint (iteration_count == 0) before failing again.
+    runner.reset_iteration_count()
+    await executor_a.execute(MockMessage(data=0), ["START"], state, ctx)
+    with pytest.raises(RuntimeError, match="Executor failed with pending events"):
+        async for _ in runner.run_until_convergence():
+            pass
+
+    checkpoints = await storage.list_checkpoints(workflow_name="test_name")
+    assert any(cp.iteration_count == 0 for cp in checkpoints), (
+        "Fresh run after a failed resumed run must create the superstep-0 checkpoint; "
+        "a leaked resume flag would have skipped it"
+    )
+
+
+async def test_runner_creates_entry_checkpoint_at_iteration_zero():
+    """A fresh run creates the entry (superstep-0) checkpoint at iteration 0 with no parent.
+
+    This is the baseline the lineage-consistency guard must preserve: when starting from iteration 0
+    with messages queued and not resumed, the entry checkpoint is created and begins a new lineage
+    (``previous_checkpoint_id is None``).
+    """
+    storage = InMemoryCheckpointStorage()
+    ctx = CheckpointingContext(storage)
+    # Terminal executor with no outgoing edges: the runner runs one superstep and converges.
+    source = MockExecutor(id="source")
+    state = State()
+    runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    assert runner._iteration == 0  # pyright: ignore[reportPrivateUsage]
+    assert runner._resumed_from_checkpoint is False  # pyright: ignore[reportPrivateUsage]
+
+    await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
+
+    async for _ in runner.run_until_convergence():
+        pass
+
+    checkpoints = await storage.list_checkpoints(workflow_name="test_name")
+    entry_checkpoints = [cp for cp in checkpoints if cp.iteration_count == 0]
+    assert len(entry_checkpoints) == 1, "A fresh run must create exactly one entry checkpoint at iteration 0"
+    assert entry_checkpoints[0].previous_checkpoint_id is None, (
+        "The entry checkpoint of a fresh run must begin a new lineage with no parent"
+    )
+
+
+async def test_runner_skips_entry_checkpoint_when_iteration_nonzero():
+    """The entry (superstep-0) checkpoint must only be created at iteration 0 to keep lineage consistent.
+
+    A re-run that did not reset the iteration count (and is not marked as resumed) must not write an
+    entry checkpoint carrying a non-zero ``iteration_count`` - doing so would place two checkpoints at
+    the same iteration in the lineage. The ``_iteration == 0`` guard suppresses the entry checkpoint in
+    this case while still allowing the normal per-superstep checkpoints to be created.
+    """
+    storage = InMemoryCheckpointStorage()
+    ctx = CheckpointingContext(storage)
+    # Terminal executor with no outgoing edges: the runner runs one superstep and converges.
+    source = MockExecutor(id="source")
+    state = State()
+    runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Simulate a re-run that kept its iteration count and is not marked as resumed.
+    runner._iteration = 5  # pyright: ignore[reportPrivateUsage]
+    assert runner._resumed_from_checkpoint is False  # pyright: ignore[reportPrivateUsage]
+
+    await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
+
+    async for _ in runner.run_until_convergence():
+        pass
+
+    checkpoints = await storage.list_checkpoints(workflow_name="test_name")
+    # No entry checkpoint at the pre-existing iteration count may be created.
+    assert all(cp.iteration_count != 5 for cp in checkpoints), (
+        "Entry checkpoint must not be created at a non-zero iteration; lineage would have a duplicate iteration"
+    )
+    # The normal post-superstep checkpoint is still created (iteration advanced to 6).
+    assert any(cp.iteration_count == 6 for cp in checkpoints)
+
+
+async def test_runner_resumed_from_iteration_zero_skips_entry_checkpoint():
+    """Resuming from an iteration-0 checkpoint must not recreate the entry checkpoint.
+
+    Here ``_iteration == 0`` is true, so the iteration guard alone would not suppress the entry
+    checkpoint; the resume flag is what prevents recreating the checkpoint that already exists at
+    iteration 0.
+    """
+    storage = InMemoryCheckpointStorage()
+    ctx = CheckpointingContext(storage)
+    source = MockExecutor(id="source")
+    state = State()
+    runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Resume from an iteration-0 checkpoint: iteration stays 0 but the run is marked as resumed.
+    resumed_checkpoint = WorkflowCheckpoint(
+        checkpoint_id="entry-cp",
+        workflow_name="test_name",
+        graph_signature_hash="test_hash",
+        iteration_count=0,
+    )
+    runner._mark_resumed(resumed_checkpoint)  # pyright: ignore[reportPrivateUsage]
+    assert runner._iteration == 0  # pyright: ignore[reportPrivateUsage]
+    assert runner._resumed_from_checkpoint is True  # pyright: ignore[reportPrivateUsage]
+
+    await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
+
+    async for _ in runner.run_until_convergence():
+        pass
+
+    # The pre-loop entry checkpoint is skipped; only the post-superstep checkpoint (iteration 1) is created,
+    # and it chains back to the resumed entry checkpoint.
+    checkpoints = sorted(
+        await storage.list_checkpoints(workflow_name="test_name"),
+        key=lambda c: c.timestamp,
+    )
+    assert all(cp.checkpoint_id != "entry-cp" for cp in checkpoints), "Resumed entry checkpoint must not be recreated"
+    assert checkpoints, "The resumed run must still create its post-superstep checkpoint"
+    assert checkpoints[0].previous_checkpoint_id == "entry-cp", (
+        "The first post-resume checkpoint must chain back to the resumed entry checkpoint"
+    )
+
+
 class SlowEventEmittingExecutor(Executor):
     """An executor that emits events with delays to test straggler event draining."""
 

python/packages/core/tests/workflow/test_runner_context.py

@@ -1,72 +0,0 @@
-# Copyright (c) Microsoft. All rights reserved.
-
-"""Tests for `InProcRunnerContext`."""
-
-import pytest
-
-from agent_framework import (
-    InProcRunnerContext,
-    WorkflowEvent,
-    WorkflowMessage,
-)
-
-
-def _make_request_info_event(request_id: str, source_executor_id: str = "executor") -> WorkflowEvent[str]:
-    return WorkflowEvent.request_info(
-        request_id=request_id,
-        source_executor_id=source_executor_id,
-        request_data="please respond",
-        response_type=str,
-    )
-
-
-class TestInProcRunnerContextResetForNewRun:
-    """Verify `reset_for_new_run` clears per-run state, including pending request_info events."""
-
-    async def test_reset_clears_pending_request_info_events(self) -> None:
-        ctx = InProcRunnerContext()
-
-        await ctx.add_request_info_event(_make_request_info_event("req-1"))
-        await ctx.add_request_info_event(_make_request_info_event("req-2"))
-
-        assert set((await ctx.get_pending_request_info_events()).keys()) == {"req-1", "req-2"}
-
-        ctx.reset_for_new_run()
-
-        assert await ctx.get_pending_request_info_events() == {}
-
-    async def test_reset_clears_pending_request_info_events_when_already_empty(self) -> None:
-        ctx = InProcRunnerContext()
-
-        assert await ctx.get_pending_request_info_events() == {}
-
-        ctx.reset_for_new_run()
-
-        assert await ctx.get_pending_request_info_events() == {}
-
-    async def test_reset_after_pending_event_blocks_response_correlation(self) -> None:
-        """After `reset_for_new_run`, prior request ids must no longer correlate to a response."""
-        ctx = InProcRunnerContext()
-        await ctx.add_request_info_event(_make_request_info_event("req-1"))
-
-        ctx.reset_for_new_run()
-
-        with pytest.raises(ValueError, match="No pending request found for request_id: req-1"):
-            await ctx.send_request_info_response("req-1", "answer")
-
-    async def test_reset_clears_messages_events_and_streaming_flag(self) -> None:
-        """Sanity-check the other state `reset_for_new_run` is documented to clear."""
-        ctx = InProcRunnerContext()
-        await ctx.send_message(WorkflowMessage(data="hello", source_id="executor"))
-        await ctx.add_event(WorkflowEvent("status", data="running"))
-        ctx.set_streaming(True)
-
-        assert await ctx.has_messages() is True
-        assert await ctx.has_events() is True
-        assert ctx.is_streaming() is True
-
-        ctx.reset_for_new_run()
-
-        assert await ctx.has_messages() is False
-        assert await ctx.has_events() is False
-        assert ctx.is_streaming() is False

python/packages/core/tests/workflow/test_workflow.py

@@ -20,6 +20,7 @@ from agent_framework import (
     Content,
     Executor,
     FileCheckpointStorage,
+    InProcRunnerContext,
     Message,
     ResponseStream,
     WorkflowBuilder,
@@ -994,6 +995,90 @@ async def test_workflow_partial_stream_does_not_clobber_successor_active_run() -
     await asyncio.sleep(0)
 
 
+async def test_workflow_stale_runtime_checkpoint_storage_not_inherited() -> None:
+    """A new run must not inherit a prior run's leftover runtime checkpoint storage.
+
+    If a run that set a runtime ``checkpoint_storage`` override is dropped before
+    its async-generator finalizer clears it, the override can linger on the
+    ``RunnerContext`` while ``_is_run_active()`` already reports False. ``run()``
+    defensively clears that stale override so a subsequent run that does not pass
+    its own ``checkpoint_storage`` does not silently checkpoint into it.
+    """
+    with tempfile.TemporaryDirectory() as temp_dir:
+        leftover_storage = FileCheckpointStorage(temp_dir)
+        executor = IncrementExecutor(id="stale_storage_exec", limit=3, increment=1)
+        workflow = WorkflowBuilder(start_executor=executor).build()
+
+        assert isinstance(workflow._runner.context, InProcRunnerContext)  # pyright: ignore[reportPrivateUsage]
+
+        # Simulate a leftover runtime override from a dropped prior run.
+        workflow._runner.context.set_runtime_checkpoint_storage(leftover_storage)  # pyright: ignore[reportPrivateUsage]
+
+        # A fresh run without its own checkpoint_storage must not use the leftover.
+        result = await workflow.run(NumberMessage(data=0))
+        assert result.get_final_state() == WorkflowRunState.IDLE
+
+        checkpoints = await leftover_storage.list_checkpoints(workflow_name=workflow.name)
+        assert checkpoints == [], "Stale runtime checkpoint storage must not be inherited by a new run"
+        assert workflow._runner.context._runtime_checkpoint_storage is None  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_workflow_partial_stream_does_not_clobber_successor_runtime_storage() -> None:
+    """A stale ``_run_core`` finalizer must not clear a successor's runtime storage.
+
+    Same GC-finalizer race as
+    ``test_workflow_partial_stream_does_not_clobber_successor_active_run`` but for the
+    runtime checkpoint storage override: the dropped run's deferred ``finally`` must
+    only clear the override if it still owns it, otherwise it wipes the successor
+    run's storage.
+    """
+    with (
+        tempfile.TemporaryDirectory() as temp_dir_a,
+        tempfile.TemporaryDirectory() as temp_dir_b,
+    ):
+        storage_a = FileCheckpointStorage(temp_dir_a)
+        storage_b = FileCheckpointStorage(temp_dir_b)
+        executor = IncrementExecutor(id="storage_finalizer_exec", limit=100, increment=1)
+        workflow = WorkflowBuilder(start_executor=executor).build()
+        context = workflow._runner.context  # pyright: ignore[reportPrivateUsage]
+
+        assert isinstance(context, InProcRunnerContext)
+
+        # Step 1: drive stream A's body to its first yield so it set storage_a.
+        stream_a = workflow.run(NumberMessage(data=0), checkpoint_storage=storage_a, stream=True)
+        aiter_a = stream_a.__aiter__()
+        await aiter_a.__anext__()
+        assert context._runtime_checkpoint_storage is storage_a  # pyright: ignore[reportPrivateUsage]
+
+        # Step 2: drop stream A; the weakref dies and async-gen close is scheduled
+        # but not run inline.
+        del stream_a
+        del aiter_a
+        gc.collect()
+
+        # Step 3: synchronously start stream B with its own storage and drive it to
+        # its first yield so it set storage_b and took ownership of the override.
+        stream_b = workflow.run(NumberMessage(data=0), checkpoint_storage=storage_b, stream=True)
+        aiter_b = stream_b.__aiter__()
+        await aiter_b.__anext__()
+        assert context._runtime_checkpoint_storage is storage_b  # pyright: ignore[reportPrivateUsage]
+
+        # Step 4: yield enough for stream A's scheduled aclose to drive its body
+        # through ``GeneratorExit`` and into its ``finally``.
+        for _ in range(5):
+            await asyncio.sleep(0)
+
+        # With the ownership guard, stream B's override survives. Without it, A's
+        # stale finalizer would have cleared it.
+        assert context._runtime_checkpoint_storage is storage_b  # pyright: ignore[reportPrivateUsage]
+
+        # Tear down stream B.
+        del stream_b
+        del aiter_b
+        gc.collect()
+        await asyncio.sleep(0)
+
+
 class _StreamingTestAgent(BaseAgent):
     """Test agent that supports both streaming and non-streaming modes."""
 
@@ -1415,85 +1500,3 @@ async def test_output_executors_filtering_with_run_responses_streaming() -> None
 
 
 # endregion
-
-
-# region Workflow.reset
-
-
-class CounterStateExecutor(Executor):
-    """Executor with local mutable state used to verify checkpoint-based reset."""
-
-    def __init__(self, id: str) -> None:
-        super().__init__(id=id)
-        self.counter = 0
-
-    @handler
-    async def handle(self, message: str, ctx: WorkflowContext[str, int]) -> None:
-        self.counter += 1
-        await ctx.yield_output(self.counter)
-
-    async def on_checkpoint_save(self) -> dict[str, Any]:
-        return {"counter": self.counter}
-
-    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
-        self.counter = int(state.get("counter", 0))
-
-
-class TestWorkflowReset:
-    """Tests for :meth:`Workflow.reset`."""
-
-    async def test_reset_restores_initial_shared_state(self) -> None:
-        """Reset clears accumulated workflow state back to the initial baseline."""
-        executor = StateTrackingExecutor(id="state_executor")
-        workflow = WorkflowBuilder(start_executor=executor).add_edge(executor, executor).build()
-
-        result1 = await workflow.run(StateTrackingMessage(data="message1", run_id="run1"))
-        assert result1.get_outputs()[0] == ["run1:message1"]
-
-        result2 = await workflow.run(StateTrackingMessage(data="message2", run_id="run2"))
-        assert result2.get_outputs()[0] == ["run1:message1", "run2:message2"]
-
-        await workflow.reset()
-
-        result3 = await workflow.run(StateTrackingMessage(data="message3", run_id="run3"))
-        assert result3.get_outputs()[0] == ["run3:message3"]
-
-    async def test_reset_restores_executor_checkpoint_state(self) -> None:
-        """Reset restores per-executor local state captured in the initial checkpoint."""
-        executor = CounterStateExecutor(id="counter_executor")
-        workflow = WorkflowBuilder(start_executor=executor).add_edge(executor, executor).build()
-
-        result1 = await workflow.run("one")
-        assert result1.get_outputs() == [1]
-
-        result2 = await workflow.run("two")
-        assert result2.get_outputs() == [2]
-
-        await workflow.reset()
-
-        result3 = await workflow.run("three")
-        assert result3.get_outputs() == [1]
-
-    async def test_reset_before_first_run_is_allowed(self, simple_executor: Executor) -> None:
-        """Reset can be called before the first run and leaves workflow runnable."""
-        workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
-
-        await workflow.reset()
-
-        result = await workflow.run("hello")
-        assert result.get_final_state() == WorkflowRunState.IDLE
-
-    async def test_reset_raises_while_run_active(self, simple_executor: Executor) -> None:
-        """Reset must reject while a workflow run is active."""
-        workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
-
-        active_stream = workflow.run(WorkflowMessage(data="hi", source_id="test"), stream=True)
-        try:
-            with pytest.raises(WorkflowException, match="Cannot reset workflow while a run is active"):
-                await workflow.reset()
-        finally:
-            async for _ in active_stream:
-                pass
-
-
-# endregion

python/packages/declarative/tests/test_http_request_yaml_integration.py

@@ -74,7 +74,7 @@ async def test_http_request_yaml_roundtrip() -> None:
     await workflow.run({})
 
     decl: dict[str, Any] = workflow._runner.state.get(DECLARATIVE_STATE_KEY) or {}
-    local = decl.get("Local") or {}
+    local: dict[str, Any] = decl.get("Local") or {}
 
     assert local.get("RepoOwner") == "dotnet"
     repo_info = local.get("RepoInfo")

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -386,6 +386,7 @@ class ResponsesHostServer(ResponsesAgentServerHost):
             )
 
         self._is_workflow_agent = False
+        self._checkpoint_storage_path = None
         if isinstance(agent, WorkflowAgent):
             if agent.workflow._runner_context.has_checkpointing():  # pyright: ignore[reportPrivateUsage]
                 raise RuntimeError(
@@ -579,6 +580,8 @@ class ResponsesHostServer(ResponsesAgentServerHost):
 
             # The following should never happen due to the checks above.
             # This is for type safety and defensive programming.
+            if self._checkpoint_storage_path is None:
+                raise RuntimeError("Checkpoint storage path is not configured for workflow agent.")
             if not isinstance(self._agent, WorkflowAgent):
                 raise RuntimeError("Agent is not a workflow agent.")
 
@@ -596,27 +599,43 @@ class ResponsesHostServer(ResponsesAgentServerHost):
             # the only place that state lives is the workflow checkpoint, so
             # on every turn we restore the latest checkpoint and feed the new
             # input back into the start executor as a continuation rather than
-            # a fresh run. If no conversation_id or previous_response_id is
-            # supplied (or no checkpoint exists for that context), reset the
-            # workflow to its in-memory initial baseline to avoid context bleed
-            # between requests.
+            # a fresh run.
             latest_checkpoint_id: str | None = None
             restore_storage: FileCheckpointStorage | None = None
             if context_id is not None:
-                context_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, context_id)
-                latest_checkpoint = await context_storage.get_latest(workflow_name=self._agent.workflow.name)
+                restore_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, context_id)
+                latest_checkpoint = await restore_storage.get_latest(workflow_name=self._agent.workflow.name)
                 if latest_checkpoint is not None:
                     latest_checkpoint_id = latest_checkpoint.checkpoint_id
-                    restore_storage = context_storage
 
-            # Restore the workflow to the latest checkpoint and run it with the
-            # new input. Events (including request info events) will not be emitted
-            # during restoration (in streaming) or after restoration (in non-streaming)
-            # since we assume the client had already seen those events and we don't want
-            # to emit duplicates.
-            if latest_checkpoint_id is None or restore_storage is None:
-                await self._agent.workflow.reset()
-            else:
+            # Storage that will receive checkpoints written during this turn.
+            # When the caller chains with previous_response_id, the next turn
+            # will reference the current response_id as its previous_response_id,
+            # so new checkpoints must land under the current response_id (or the
+            # conversation_id when set). When conversation_id is set, this
+            # matches restore_storage; when only previous_response_id was
+            # supplied, restore_storage points at the *prior* response's
+            # directory and write_storage points at the *current* response's.
+            write_context_id = context.conversation_id or context.response_id
+            write_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, write_context_id)
+
+            # Multi-turn pattern: when we have a prior checkpoint, restore it
+            # first (drive the workflow back to idle with prior state intact),
+            # then make a separate call that delivers the new user input. This
+            # depends on Workflow.run preserving shared state across calls. The
+            # restore-only call may yield events from any pending in-flight
+            # work in the checkpoint; we consume those internally here so they
+            # don't surface to the response stream as duplicates.
+            #
+            # If the restored checkpoint had pending request_info events, the
+            # restore-only call replays them through
+            # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``
+            # and populates ``self._agent.pending_requests``. That is the correct
+            # state: those requests are genuinely outstanding, and the next
+            # ``run(input_messages, ...)`` call may contain ``function_call_output``
+            # items (carried as FunctionResult/FunctionApprovalResponse content)
+            # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`.
+            if latest_checkpoint_id is not None:
                 if is_streaming_request:
                     async for _ in self._agent.run(
                         stream=True,
@@ -631,17 +650,6 @@ class ResponsesHostServer(ResponsesAgentServerHost):
                         checkpoint_storage=restore_storage,
                     )
 
-            # Storage that will receive checkpoints written during this turn.
-            # When the caller chains with previous_response_id, the next turn
-            # will reference the current response_id as its previous_response_id,
-            # so new checkpoints must land under the current response_id (or the
-            # conversation_id when set). When conversation_id is set, this
-            # matches restore_storage; when only previous_response_id was
-            # supplied, restore_storage points at the *prior* response's
-            # directory and write_storage points at the *current* response's.
-            write_context_id = context.conversation_id or context.response_id
-            write_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, write_context_id)
-
             if not is_streaming_request:
                 # Run the agent in non-streaming mode with the new user input.
                 response = await self._agent.run(

python/packages/foundry_hosting/tests/test_responses.py

@@ -3062,7 +3062,6 @@ class TestCheckpointContextPathValidation:
         agent.workflow = MagicMock()
         agent.workflow.name = "wf"
         agent.workflow._runner_context.has_checkpointing = MagicMock(return_value=False)
-        agent.workflow.reset = AsyncMock()
         agent.run = AsyncMock(
             side_effect=[
                 AgentResponse(messages=[]),
@@ -3093,136 +3092,6 @@ class TestCheckpointContextPathValidation:
         assert new_turn_messages[0].text == "next turn"
         assert new_turn_call.kwargs["checkpoint_storage"].storage_path == (root / response_id).resolve()
 
-    async def test_handle_inner_workflow_resets_when_no_context_id(self, tmp_path: Any) -> None:
-        """When no context id is supplied, the workflow resets to its initial in-memory state."""
-        from agent_framework import WorkflowAgent
-        from azure.ai.agentserver.responses import ResponseContext
-        from azure.ai.agentserver.responses.models import CreateResponse, ItemMessage
-
-        response_id = "resp_current"
-        root = tmp_path / "root"
-        root.mkdir()
-
-        agent = MagicMock(spec=WorkflowAgent)
-        agent.id = "wf-agent"
-        agent.name = "wf"
-        agent.description = ""
-        agent.context_providers = []
-        agent.workflow = MagicMock()
-        agent.workflow.name = "wf"
-        agent.workflow._runner_context.has_checkpointing = MagicMock(return_value=False)
-        agent.workflow.reset = AsyncMock()
-        agent.run = AsyncMock(
-            return_value=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
-        )
-        server = ResponsesHostServer(agent, store=InMemoryResponseProvider())
-        server._checkpoint_storage_path = str(root)  # pyright: ignore[reportPrivateUsage]
-
-        # No previous_response_id and no conversation_id.
-        request = CreateResponse(model="m", input="hi")
-        context = ResponseContext(response_id=response_id, mode_flags=MagicMock())
-        input_item = ItemMessage({"type": "message", "role": "user", "content": "fresh turn"})
-
-        with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[input_item])):
-            async for _ in server._handle_inner_workflow(request, context):  # pyright: ignore[reportPrivateUsage]
-                pass
-
-        # No checkpoint restore is attempted; workflow resets in memory.
-        assert agent.workflow.reset.await_count == 1
-        assert agent.run.call_count == 1
-
-        # The single run() call delivers the new input; checkpoints land under response_id
-        # (the write-sink directory keyed by the current response id).
-        new_turn_call = agent.run.call_args_list[0]
-        new_turn_messages = new_turn_call.args[0]
-        assert len(new_turn_messages) == 1
-        assert new_turn_messages[0].text == "fresh turn"
-        assert new_turn_call.kwargs["checkpoint_storage"].storage_path == (root / response_id).resolve()
-
-    async def test_handle_inner_workflow_resets_each_request_without_context_id(self, tmp_path: Any) -> None:
-        """Requests without context ids reset workflow state per request."""
-        from agent_framework import WorkflowAgent
-        from azure.ai.agentserver.responses import ResponseContext
-        from azure.ai.agentserver.responses.models import CreateResponse, ItemMessage
-
-        root = tmp_path / "root"
-        root.mkdir()
-
-        agent = MagicMock(spec=WorkflowAgent)
-        agent.id = "wf-agent"
-        agent.name = "wf"
-        agent.description = ""
-        agent.context_providers = []
-        agent.workflow = MagicMock()
-        agent.workflow.name = "wf"
-        agent.workflow._runner_context.has_checkpointing = MagicMock(return_value=False)
-        agent.workflow.reset = AsyncMock()
-        # Two run() calls total: one new turn per request.
-        agent.run = AsyncMock(return_value=AgentResponse(messages=[]))
-
-        server = ResponsesHostServer(agent, store=InMemoryResponseProvider())
-        server._checkpoint_storage_path = str(root)  # pyright: ignore[reportPrivateUsage]
-
-        request1 = CreateResponse(model="m", input="hi")
-        context1 = ResponseContext(response_id="resp_first", mode_flags=MagicMock())
-        request2 = CreateResponse(model="m", input="hi again")
-        context2 = ResponseContext(response_id="resp_second", mode_flags=MagicMock())
-        input_item = ItemMessage({"type": "message", "role": "user", "content": "turn"})
-
-        with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[input_item])):
-            async for _ in server._handle_inner_workflow(request1, context1):  # pyright: ignore[reportPrivateUsage]
-                pass
-            async for _ in server._handle_inner_workflow(request2, context2):  # pyright: ignore[reportPrivateUsage]
-                pass
-
-        assert agent.workflow.reset.await_count == 2
-        assert agent.run.call_count == 2
-
-    async def test_handle_inner_workflow_resets_when_context_dir_is_empty(self, tmp_path: Any) -> None:
-        """When previous_response_id has no checkpoint, workflow resets instead of restoring."""
-        from agent_framework import WorkflowAgent
-        from azure.ai.agentserver.responses import ResponseContext
-        from azure.ai.agentserver.responses.models import CreateResponse, ItemMessage
-
-        previous_response_id = "resp_previous"
-        response_id = "resp_current"
-        root = tmp_path / "root"
-        root.mkdir()
-        # The per-context storage exists but contains no checkpoints.
-        (root / previous_response_id).mkdir()
-
-        agent = MagicMock(spec=WorkflowAgent)
-        agent.id = "wf-agent"
-        agent.name = "wf"
-        agent.description = ""
-        agent.context_providers = []
-        agent.workflow = MagicMock()
-        agent.workflow.name = "wf"
-        agent.workflow._runner_context.has_checkpointing = MagicMock(return_value=False)
-        agent.workflow.reset = AsyncMock()
-        agent.run = AsyncMock(
-            return_value=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
-        )
-        server = ResponsesHostServer(agent, store=InMemoryResponseProvider())
-        server._checkpoint_storage_path = str(root)  # pyright: ignore[reportPrivateUsage]
-
-        request = CreateResponse(model="m", input="hi", previous_response_id=previous_response_id)
-        context = ResponseContext(
-            response_id=response_id, previous_response_id=previous_response_id, mode_flags=MagicMock()
-        )
-        input_item = ItemMessage({"type": "message", "role": "user", "content": "next turn"})
-
-        with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[input_item])):
-            async for _ in server._handle_inner_workflow(request, context):  # pyright: ignore[reportPrivateUsage]
-                pass
-
-        assert agent.workflow.reset.await_count == 1
-        assert agent.run.call_count == 1
-
-        # The new turn writes checkpoints under the current response id.
-        new_turn_call = agent.run.call_args_list[0]
-        assert new_turn_call.kwargs["checkpoint_storage"].storage_path == (root / response_id).resolve()
-
     @pytest.mark.parametrize(
         "bad_id",
         [
@@ -3316,8 +3185,6 @@ class TestCheckpointContextPathValidation:
         agent.workflow = MagicMock()
         agent.workflow.name = "wf"
         agent.workflow._runner_context.has_checkpointing = MagicMock(return_value=False)
-        agent.workflow.reset = AsyncMock()
-        agent.run = AsyncMock(return_value=AgentResponse(messages=[]))
 
         # Constructor inspects WorkflowAgent.workflow internals; bypass setup
         # by feeding a configured mock through a normal init.

python/packages/hosting-responses/LICENSE

@@ -0,0 +1,21 @@
+    MIT License
+
+    Copyright (c) Microsoft Corporation.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"), to deal
+    in the Software without restriction, including without limitation the rights
+    to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+    copies of the Software, and to permit persons to whom the Software is
+    furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all
+    copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+    AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+    OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+    SOFTWARE

python/packages/hosting-responses/README.md

@@ -0,0 +1,21 @@
+# agent-framework-hosting-responses
+
+OpenAI Responses-shaped channel for `agent-framework-hosting`.
+
+Exposes a single `POST /responses` endpoint that accepts the OpenAI
+Responses API request body and returns either a Responses-shaped JSON
+body or a Server-Sent-Events stream when `stream=True`.
+
+```python
+from agent_framework.openai import OpenAIChatClient
+from agent_framework_hosting import AgentFrameworkHost
+from agent_framework_hosting_responses import ResponsesChannel
+
+agent = OpenAIChatClient().as_agent(name="Assistant")
+
+host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel()])
+host.serve(port=8000)
+```
+
+The base host plumbing lives in
+[`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/).

python/packages/hosting-responses/agent_framework_hosting_responses/__init__.py

@@ -0,0 +1,25 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""OpenAI Responses-shaped channel for ``agent-framework-hosting``."""
+
+import importlib.metadata
+
+from ._channel import ResponsesChannel
+from ._parsing import (
+    messages_from_responses_input,
+    parse_responses_identity,
+    parse_responses_request,
+)
+
+try:
+    __version__ = importlib.metadata.version(__name__)
+except importlib.metadata.PackageNotFoundError:
+    __version__ = "0.0.0"
+
+__all__ = [
+    "ResponsesChannel",
+    "__version__",
+    "messages_from_responses_input",
+    "parse_responses_identity",
+    "parse_responses_request",
+]

python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py

@@ -0,0 +1,156 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Parsing helpers for the OpenAI Responses-API request body.
+
+The Responses API accepts ``input`` as either a string or a list of "input
+items". An item is either a content part (``input_text`` / ``input_image``
+/ ``input_file``) or a message envelope ``{type: "message", role,
+content: [...]}``. We translate that into an Agent Framework ``Message``
+list and remap the generation-control fields the API also carries into
+``ChatOptions``-shaped keys. The result is available to the channel's
+``run_hook``; a default hook strips them before they reach the agent so
+unknown fields from untrusted callers are not forwarded unless the host
+developer explicitly opts in.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, cast
+
+from agent_framework import Content, Message
+from agent_framework_hosting import ChannelIdentity, ChannelSession
+
+# OpenAI Responses field name → Agent Framework ChatOptions field name.
+_RESPONSES_OPTION_REMAP = {
+    "max_output_tokens": "max_tokens",
+    "parallel_tool_calls": "allow_multiple_tool_calls",
+}
+# Fields the Responses transport owns; they are consumed separately and must
+# not also appear in options.
+_RESPONSES_TRANSPORT_KEYS = frozenset({"input", "stream", "previous_response_id"})
+
+
+def parse_responses_identity(body: Mapping[str, Any], channel_name: str) -> ChannelIdentity | None:
+    """Surface the caller as a :class:`ChannelIdentity` so the host can record it.
+
+    OpenAI Responses replaced ``user`` with ``safety_identifier`` — we use
+    that as the native id, falling back to the legacy ``user`` field.
+    """
+    native = body.get("safety_identifier") or body.get("user")
+    if not isinstance(native, str) or not native:
+        return None
+    return ChannelIdentity(channel=channel_name, native_id=native)
+
+
+def _content_from_input_item(item: Mapping[str, Any]) -> Content:
+    """Convert a single OpenAI Responses ``input`` item into a :class:`Content` part.
+
+    Handles the ``input_text``/``output_text``/``text`` text variants,
+    ``input_image`` URL references, and ``input_file`` references via either
+    a public URL or a hosted ``file_id``. Raises ``ValueError`` for any
+    unsupported item type so the surrounding parser can return a 422.
+    """
+    item_type = item.get("type")
+    if item_type in ("input_text", "output_text", "text"):
+        return Content.from_text(text=str(item.get("text", "")))
+    if item_type == "input_image":
+        image_url: Any = item.get("image_url")
+        if isinstance(image_url, Mapping):
+            image_url = cast("Mapping[str, Any]", image_url).get("url")
+        if not isinstance(image_url, str):
+            raise ValueError("input_image requires `image_url`")
+        return Content.from_uri(uri=image_url, media_type="image/*")
+    if item_type == "input_file":
+        if (uri := item.get("file_url")) and isinstance(uri, str):
+            return Content.from_uri(uri=uri, media_type=item.get("mime_type"))
+        if file_id := item.get("file_id"):
+            return Content(type="hosted_file", file_id=str(file_id))
+        raise ValueError("input_file requires `file_url` or `file_id`")
+    raise ValueError(f"Unsupported Responses input content type: {item_type!r}")
+
+
+def messages_from_responses_input(value: Any) -> list[Message]:
+    """Translate ``input`` (string or list of items) into :class:`Message` objects."""
+    if isinstance(value, str):
+        return [Message("user", [Content.from_text(text=value)])]
+    if not isinstance(value, list) or not value:
+        raise ValueError("`input` must be a non-empty string or list")
+
+    messages: list[Message] = []
+    pending_user_parts: list[Content] = []
+
+    def flush() -> None:
+        """Emit any buffered loose user content as a single user message."""
+        if pending_user_parts:
+            messages.append(Message("user", list(pending_user_parts)))
+            pending_user_parts.clear()
+
+    for item in cast("list[Any]", value):
+        if not isinstance(item, Mapping):
+            raise ValueError("each `input` item must be an object")
+        item_map = cast("Mapping[str, Any]", item)
+        if item_map.get("type") == "message":
+            flush()
+            role = str(item_map.get("role") or "user")
+            content: Any = item_map.get("content") or []
+            parts: list[Content]
+            if isinstance(content, str):
+                parts = [Content.from_text(text=content)]
+            elif isinstance(content, list):
+                parts = []
+                for content_item in cast("list[Any]", content):
+                    if not isinstance(content_item, Mapping):
+                        raise ValueError("each message `content` item must be an object")
+                    parts.append(_content_from_input_item(cast("Mapping[str, Any]", content_item)))
+            else:
+                raise ValueError("message `content` must be a string or list")
+            messages.append(Message(role, parts))
+        else:
+            pending_user_parts.append(_content_from_input_item(item_map))
+
+    flush()
+    if not messages:
+        raise ValueError("`input` produced no messages")
+    return messages
+
+
+def parse_responses_request(
+    body: Mapping[str, Any],
+) -> tuple[list[Message], dict[str, Any], ChannelSession | None]:
+    """Translate a Responses-API request body into Agent Framework constructs.
+
+    Returns a triple ``(messages, options, session)`` where:
+
+    - ``messages`` is the parsed conversation.
+    - ``options`` is a ``ChatOptions``-shaped dict with the remapped
+      generation-control fields. Known Responses→ChatOptions renames are
+      applied (e.g. ``max_output_tokens`` → ``max_tokens``); transport/
+      session keys are excluded; ``None``-valued fields are dropped.
+      Unknown fields are forwarded as-is so the channel's ``run_hook``
+      can inspect and filter them. The default ``ResponsesChannel`` strips
+      all options before the agent runs; supply a custom ``run_hook`` to
+      selectively keep fields.
+    - ``session`` is a :class:`ChannelSession` keyed by
+      ``previous_response_id`` when one was supplied, else ``None``.
+    """
+    messages = messages_from_responses_input(body.get("input"))
+
+    options: dict[str, Any] = {}
+    for key, value in body.items():
+        if key in _RESPONSES_TRANSPORT_KEYS or value is None:
+            continue
+        options[_RESPONSES_OPTION_REMAP.get(key, key)] = value
+
+    session: ChannelSession | None = None
+    if (prev := body.get("previous_response_id")) and isinstance(prev, str):
+        session = ChannelSession(isolation_key=prev)
+
+    return messages, options, session
+
+
+__all__ = [
+    "messages_from_responses_input",
+    "parse_responses_identity",
+    "parse_responses_request",
+]

python/packages/hosting-responses/pyproject.toml

@@ -0,0 +1,80 @@
+[project]
+name = "agent-framework-hosting-responses"
+description = "OpenAI Responses-shaped channel for agent-framework-hosting."
+authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
+readme = "README.md"
+requires-python = ">=3.10"
+version = "1.0.0a260424"
+license-files = ["LICENSE"]
+urls.homepage = "https://aka.ms/agent-framework"
+urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
+urls.issues = "https://github.com/microsoft/agent-framework/issues"
+classifiers = [
+  "License :: OSI Approved :: MIT License",
+  "Development Status :: 3 - Alpha",
+  "Intended Audience :: Developers",
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
+  "Typing :: Typed",
+]
+dependencies = [
+    "agent-framework-core>=1.2.0,<2",
+    "agent-framework-hosting==1.0.0a260424",
+    "openai>=1.99.0,<3",
+]
+
+[tool.uv]
+prerelease = "if-necessary-or-explicit"
+environments = [
+    "sys_platform == 'darwin'",
+    "sys_platform == 'linux'",
+    "sys_platform == 'win32'"
+]
+
+[tool.uv-dynamic-versioning]
+fallback-version = "0.0.0"
+
+[tool.pytest.ini_options]
+testpaths = 'tests'
+addopts = "-ra -q -r fEX"
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+filterwarnings = []
+timeout = 120
+markers = [
+    "integration: marks tests as integration tests that require external services",
+]
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+
+[tool.coverage.run]
+omit = [
+    "**/__init__.py"
+]
+
+[tool.pyright]
+extends = "../../pyproject.toml"
+include = ["agent_framework_hosting_responses"]
+exclude = ['tests']
+
+[tool.bandit]
+targets = ["agent_framework_hosting_responses"]
+exclude_dirs = ["tests"]
+
+[tool.poe]
+executor.type = "uv"
+include = "../../shared_tasks.toml"
+
+[tool.poe.tasks.test]
+help = "Run the default unit test suite for this package."
+cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_responses --cov-report=term-missing:skip-covered tests'
+
+[build-system]
+requires = ["flit-core >= 3.11,<4.0"]
+build-backend = "flit_core.buildapi"

python/packages/hosting-responses/tests/hosting_responses/test_channel.py

@@ -0,0 +1,651 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""End-to-end tests for :class:`ResponsesChannel` via Starlette's ``TestClient``."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Any
+
+from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message
+from agent_framework_hosting import (
+    AgentFrameworkHost,
+    HostedRunResult,
+)
+from starlette.testclient import TestClient
+
+from agent_framework_hosting_responses import ResponsesChannel
+from agent_framework_hosting_responses._channel import (  # pyright: ignore[reportPrivateUsage]
+    _result_to_output_items,
+    _result_to_text,
+)
+
+# --------------------------------------------------------------------------- #
+# Fakes                                                                        #
+# --------------------------------------------------------------------------- #
+
+
+@dataclass
+class _FakeAgentResponse:
+    text: str
+
+
+class _FakeStream:
+    """Minimal stand-in for AF's ``ResponseStream`` returned by ``run(stream=True)``."""
+
+    def __init__(self, chunks: list[str]) -> None:
+        self._chunks = chunks
+        self._final = _FakeAgentResponse(text="".join(chunks))
+
+    def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
+        async def _gen() -> AsyncIterator[AgentResponseUpdate]:
+            for c in self._chunks:
+                yield AgentResponseUpdate(contents=[Content.from_text(c)], role="assistant")
+
+        return _gen()
+
+    async def get_final_response(self) -> _FakeAgentResponse:
+        return self._final
+
+
+class _FakeAgent:
+    def __init__(self, reply: Any = "hello", chunks: list[str] | None = None) -> None:
+        self.id = "fake-agent"
+        self.name: str | None = "Fake Agent"
+        self.description: str | None = "Test fake agent"
+        self._reply = reply
+        self._chunks = chunks or [reply]
+        self.calls: list[dict[str, Any]] = []
+
+    def create_session(self, *, session_id: str | None = None) -> Any:
+        return {"session_id": session_id}
+
+    def get_session(self, service_session_id: str, *, session_id: str | None = None) -> Any:
+        return {"service_session_id": service_session_id, "session_id": session_id}
+
+    def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
+        self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
+        if stream:
+            return _FakeStream(self._chunks)
+
+        async def _coro() -> Any:
+            if not isinstance(self._reply, str):
+                return self._reply
+            return _FakeAgentResponse(text=self._reply)
+
+        return _coro()
+
+
+# --------------------------------------------------------------------------- #
+# Tests                                                                        #
+# --------------------------------------------------------------------------- #
+
+
+def _make_client(
+    agent: _FakeAgent | None = None,
+    *,
+    path: str = "/responses",
+    response_id_factory: Any | None = None,
+) -> tuple[TestClient, AgentFrameworkHost, _FakeAgent]:
+    agent = agent or _FakeAgent()
+    host = AgentFrameworkHost(
+        target=agent,
+        channels=[ResponsesChannel(path=path, response_id_factory=response_id_factory)],
+    )
+    return TestClient(host.app), host, agent
+
+
+def _sse_payload(body: str, event_type: str) -> dict[str, Any]:
+    current_event: str | None = None
+    for line in body.splitlines():
+        if line.startswith("event: "):
+            current_event = line[len("event: ") :]
+            continue
+        if current_event == event_type and line.startswith("data: "):
+            return json.loads(line[len("data: ") :])
+    raise AssertionError(f"Missing SSE event: {event_type}")
+
+
+class TestResponsesChannelNonStreaming:
+    def test_post_responses_returns_completed_envelope(self) -> None:
+        client, _host, agent = _make_client(_FakeAgent(reply="hi back"))
+        with client:
+            r = client.post("/responses", json={"input": "hi"})
+        assert r.status_code == 200
+        body = r.json()
+        assert body["status"] == "completed"
+        assert body["object"] == "response"
+        assert body["id"].startswith("resp_")
+        assert isinstance(body["created_at"], int)
+        assert body["output"][0]["content"][0]["text"] == "hi back"
+        assert len(agent.calls) == 1
+
+    def test_non_string_model_falls_back_to_agent(self) -> None:
+        client, _host, _agent = _make_client(_FakeAgent(reply="hi"))
+        with client:
+            r = client.post("/responses", json={"input": "hi", "model": None})
+        assert r.status_code == 200
+        assert r.json()["model"] == "agent"
+
+    def test_empty_path_mounts_at_app_root(self) -> None:
+        client, _host, _agent = _make_client(_FakeAgent(reply="hi back"), path="")
+        with client:
+            r = client.post("/", json={"input": "hi"})
+        assert r.status_code == 200
+        assert r.json()["output"][0]["content"][0]["text"] == "hi back"
+
+    def test_custom_path_mounts_route_under_host_path(self) -> None:
+        client, _host, _agent = _make_client(_FakeAgent(reply="custom"), path="/api/responses")
+        with client:
+            r = client.post("/api/responses", json={"input": "hi"})
+            missing = client.post("/api/responses/responses", json={"input": "hi"})
+        assert r.status_code == 200
+        assert r.json()["output"][0]["content"][0]["text"] == "custom"
+        assert missing.status_code == 404
+
+    def test_invalid_json_returns_400(self) -> None:
+        client, *_ = _make_client()
+        with client:
+            r = client.post("/responses", content=b"{not json", headers={"content-type": "application/json"})
+        assert r.status_code == 400
+
+    def test_non_object_json_returns_422(self) -> None:
+        client, *_ = _make_client()
+        with client:
+            r = client.post("/responses", json=["not", "an", "object"])
+        assert r.status_code == 422
+        assert r.json()["error"] == "request body must be a JSON object"
+
+    def test_invalid_input_returns_422(self) -> None:
+        client, *_ = _make_client()
+        with client:
+            r = client.post("/responses", json={"input": 42})
+        assert r.status_code == 422
+
+    def test_request_options_are_not_forwarded_by_default(self) -> None:
+        client, _host, agent = _make_client()
+        with client:
+            r = client.post(
+                "/responses",
+                json={"input": "x", "temperature": 0.5, "max_output_tokens": 64, "truncation": "auto"},
+            )
+        assert r.status_code == 200
+        assert "options" not in agent.calls[0]["kwargs"]
+
+    def test_custom_run_hook_can_forward_options(self) -> None:
+        import dataclasses
+
+        def keep_temperature(request: Any, **_: Any) -> Any:
+            opts = dict(request.options or {})
+            return dataclasses.replace(request, options={"temperature": opts.get("temperature")})
+
+        agent = _FakeAgent()
+        host = AgentFrameworkHost(
+            target=agent,
+            channels=[ResponsesChannel(run_hook=keep_temperature)],
+        )
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "x", "temperature": 0.7, "truncation": "auto"})
+        assert r.status_code == 200
+        opts = agent.calls[0]["kwargs"]["options"]
+        assert opts == {"temperature": 0.7}
+        assert "truncation" not in opts
+
+    def test_multimodal_agent_response_outputs_are_preserved(self) -> None:
+        response = AgentResponse(
+            messages=[
+                Message(
+                    "assistant",
+                    [
+                        Content.from_text_reasoning(id="rs_1", text="checking"),
+                        Content.from_function_call("call_1", "collect_media", arguments={"city": "Seattle"}),
+                        Content.from_function_result(
+                            "call_1",
+                            result=[
+                                Content.from_text("caption"),
+                                Content.from_uri("https://example.com/cat.png", media_type="image/png"),
+                                Content.from_hosted_file("file_pdf", media_type="application/pdf"),
+                            ],
+                        ),
+                        Content.from_text("done"),
+                    ],
+                )
+            ],
+        )
+        client, _host, _agent = _make_client(_FakeAgent(reply=response))
+
+        with client:
+            r = client.post("/responses", json={"input": "hi"})
+
+        assert r.status_code == 200
+        output = r.json()["output"]
+        assert [item["type"] for item in output] == [
+            "reasoning",
+            "function_call",
+            "function_call_output",
+            "message",
+        ]
+        assert output[0]["content"][0]["text"] == "checking"
+        assert output[1]["name"] == "collect_media"
+        assert output[1]["arguments"] == '{"city": "Seattle"}'
+        assert output[2]["output"] == [
+            {"text": "caption", "type": "input_text"},
+            {"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"},
+            {"type": "input_file", "file_id": "file_pdf"},
+        ]
+        assert output[3]["content"][0]["text"] == "done"
+
+    def test_raw_responses_output_items_are_preserved(self) -> None:
+        raw_item = {
+            "id": "ig_1",
+            "type": "image_generation_call",
+            "result": "base64-image",
+            "status": "completed",
+        }
+        response = AgentResponse(
+            messages=[
+                Message(
+                    "assistant",
+                    [
+                        Content.from_image_generation_tool_call(image_id="ig_1", raw_representation=raw_item),
+                        Content.from_image_generation_tool_result(
+                            image_id="ig_1",
+                            outputs=Content.from_uri("data:image/png;base64,base64-image", media_type="image/png"),
+                            raw_representation=raw_item,
+                        ),
+                    ],
+                )
+            ],
+        )
+        client, _host, _agent = _make_client(_FakeAgent(reply=response))
+
+        with client:
+            r = client.post("/responses", json={"input": "hi"})
+
+        assert r.status_code == 200
+        assert r.json()["output"] == [raw_item]
+
+    def test_later_raw_responses_output_item_replaces_earlier_partial_item(self) -> None:
+        partial = {
+            "id": "mcp_1",
+            "type": "mcp_call",
+            "server_label": "weather",
+            "name": "lookup",
+            "arguments": "{}",
+            "status": "in_progress",
+        }
+        completed = {**partial, "status": "completed", "output": "sunny"}
+        response = AgentResponse(
+            messages=[
+                Message(
+                    "assistant",
+                    [
+                        Content.from_mcp_server_tool_call(
+                            "mcp_1",
+                            "lookup",
+                            server_name="weather",
+                            raw_representation=partial,
+                        ),
+                        Content.from_mcp_server_tool_result("mcp_1", output="sunny", raw_representation=completed),
+                    ],
+                )
+            ],
+        )
+        client, _host, _agent = _make_client(_FakeAgent(reply=response))
+
+        with client:
+            r = client.post("/responses", json={"input": "hi"})
+
+        assert r.status_code == 200
+        assert r.json()["output"] == [completed]
+
+    def test_previous_response_id_creates_session(self) -> None:
+        client, _host, agent = _make_client()
+        with client:
+            client.post("/responses", json={"input": "x", "previous_response_id": "resp_42"})
+        # AgentFrameworkHost converts the channel session into an AgentSession.
+        sess = agent.calls[0]["kwargs"].get("session")
+        assert sess is not None
+        # _FakeAgent.create_session stashes the session_id on the dict it returns.
+        assert sess["session_id"] == "resp_42"
+
+    def test_first_turn_response_id_creates_session(self) -> None:
+        client, _host, agent = _make_client(response_id_factory=lambda *_: "resp_first")
+        with client:
+            client.post("/responses", json={"input": "x"})
+        sess = agent.calls[0]["kwargs"].get("session")
+        assert sess is not None
+        assert sess["session_id"] == "resp_first"
+
+    def test_chat_isolation_header_ignored_outside_foundry(self) -> None:
+        client, _host, agent = _make_client(response_id_factory=lambda *_: "resp_local")
+        with client:
+            client.post(
+                "/responses",
+                json={"input": "x"},
+                headers={"x-agent-chat-isolation-key": "chat-abc"},
+            )
+        sess = agent.calls[0]["kwargs"].get("session")
+        assert sess is not None
+        assert sess["session_id"] == "resp_local"
+
+    def test_chat_isolation_header_creates_session_in_foundry(self, monkeypatch: Any) -> None:
+        """Foundry-style ``x-agent-chat-isolation-key`` falls back to a session anchor.
+
+        First-turn requests have no ``previous_response_id`` (the client
+        doesn't have one yet), but Foundry Hosted Agents always inject
+        the isolation headers. The channel must derive a session from the
+        chat key so the host can build a stable per-conversation session
+        that history providers persist under.
+        """
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        client, _host, agent = _make_client()
+        with client:
+            client.post(
+                "/responses",
+                json={"input": "x"},
+                headers={"x-agent-chat-isolation-key": "chat-abc"},
+            )
+        sess = agent.calls[0]["kwargs"].get("session")
+        assert sess is not None
+        assert sess["session_id"] == "chat-abc"
+
+    def test_prev_response_id_wins_over_chat_isolation_header(self, monkeypatch: Any) -> None:
+        """When both anchors are present, ``previous_response_id`` wins.
+
+        ``previous_response_id`` is the protocol-native chain anchor; the
+        header fallback is only meant to bootstrap when no protocol
+        anchor exists.
+        """
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        client, _host, agent = _make_client()
+        with client:
+            client.post(
+                "/responses",
+                json={"input": "x", "previous_response_id": "resp_99"},
+                headers={"x-agent-chat-isolation-key": "chat-abc"},
+            )
+        sess = agent.calls[0]["kwargs"].get("session")
+        assert sess is not None
+        assert sess["session_id"] == "resp_99"
+
+    def test_response_hook_can_rewrite_originating_reply(self) -> None:
+        seen_kwargs: list[dict[str, Any]] = []
+
+        def hook(result: HostedRunResult, **kwargs: Any) -> HostedRunResult:
+            seen_kwargs.append(dict(kwargs))
+            return HostedRunResult(_FakeAgentResponse(text=result.result.text.upper()), session=result.session)
+
+        agent = _FakeAgent(reply="hooked")
+        host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel(response_hook=hook)])
+
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "hi"})
+
+        assert r.status_code == 200
+        body = r.json()
+        assert body["output"][0]["content"][0]["text"] == "HOOKED"
+        assert seen_kwargs
+        assert seen_kwargs[0]["channel_name"] == "responses"
+
+
+class TestResultTextRendering:
+    def test_result_text_prefers_text_property(self) -> None:
+        assert _result_to_text(_FakeAgentResponse(text="plain")) == "plain"
+
+    def test_result_text_projects_workflow_outputs(self) -> None:
+        class _WorkflowResult:
+            def get_outputs(self) -> list[Any]:
+                return [_FakeAgentResponse(text="one"), " two"]
+
+        assert _result_to_text(_WorkflowResult()) == "one two"
+
+    def test_result_output_items_project_workflow_message_and_content_outputs(self) -> None:
+        class _WorkflowResult:
+            def get_outputs(self) -> list[Any]:
+                return [
+                    Message("assistant", [Content.from_text("one")]),
+                    Content.from_function_result(
+                        "call_1",
+                        result=[Content.from_uri("https://example.com/cat.png", media_type="image/png")],
+                    ),
+                ]
+
+        output = [
+            item.model_dump(mode="json", exclude_none=True)
+            for item in _result_to_output_items(_WorkflowResult(), status="completed")
+        ]
+        assert output[0]["type"] == "message"
+        assert output[0]["content"][0]["text"] == "one"
+        assert output[1]["type"] == "function_call_output"
+        assert output[1]["output"] == [
+            {"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
+        ]
+
+    def test_function_result_exception_is_preserved(self) -> None:
+        output = [
+            item.model_dump(mode="json", exclude_none=True)
+            for item in _result_to_output_items(
+                Content.from_function_result("call_1", exception="tool failed"),
+                status="completed",
+            )
+        ]
+        assert output[0]["output"] == "tool failed"
+
+    def test_stateful_call_and_result_content_coalesce_to_one_output_item(self) -> None:
+        output = [
+            item.model_dump(mode="json", exclude_none=True)
+            for item in _result_to_output_items(
+                Message(
+                    "assistant",
+                    [
+                        Content.from_image_generation_tool_call(image_id="ig_1"),
+                        Content.from_image_generation_tool_result(
+                            image_id="ig_1",
+                            outputs=Content.from_uri("data:image/png;base64,base64-image", media_type="image/png"),
+                        ),
+                        Content.from_mcp_server_tool_call(
+                            "mcp_1",
+                            "lookup",
+                            server_name="weather",
+                            arguments={"city": "Seattle"},
+                        ),
+                        Content.from_mcp_server_tool_result("mcp_1", output=[Content.from_text("sunny")]),
+                    ],
+                ),
+                status="completed",
+            )
+        ]
+        assert output == [
+            {
+                "id": "ig_1",
+                "result": "base64-image",
+                "status": "completed",
+                "type": "image_generation_call",
+            },
+            {
+                "id": "mcp_1",
+                "arguments": '{"city": "Seattle"}',
+                "name": "lookup",
+                "output": "sunny",
+                "server_label": "weather",
+                "status": "completed",
+                "type": "mcp_call",
+            },
+        ]
+
+    def test_stateful_call_and_result_content_coalesce_across_messages(self) -> None:
+        output = [
+            item.model_dump(mode="json", exclude_none=True)
+            for item in _result_to_output_items(
+                AgentResponse(
+                    messages=[
+                        Message(
+                            "assistant",
+                            [
+                                Content.from_mcp_server_tool_call(
+                                    "mcp_1",
+                                    "lookup",
+                                    server_name="weather",
+                                    arguments={"city": "Seattle"},
+                                )
+                            ],
+                        ),
+                        Message(
+                            "tool",
+                            [Content.from_mcp_server_tool_result("mcp_1", output=[Content.from_text("sunny")])],
+                        ),
+                    ]
+                ),
+                status="completed",
+            )
+        ]
+        assert output == [
+            {
+                "id": "mcp_1",
+                "arguments": '{"city": "Seattle"}',
+                "name": "lookup",
+                "output": "sunny",
+                "server_label": "weather",
+                "status": "completed",
+                "type": "mcp_call",
+            }
+        ]
+
+
+class TestResponsesChannelStreaming:
+    def test_sse_emits_created_delta_completed(self) -> None:
+        agent = _FakeAgent(reply="hello world", chunks=["hello", " ", "world"])
+        host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel()])
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "hi", "stream": True})
+            assert r.status_code == 200
+            body = r.text
+
+        # SSE event lines look like "event: <type>\ndata: <json>\n\n".
+        events = [line[len("event: ") :] for line in body.splitlines() if line.startswith("event: ")]
+        assert events[0] == "response.created"
+        assert events[-1] == "response.completed"
+        assert events.count("response.output_text.delta") == 3
+
+    def test_sse_transform_hook_can_rewrite_chunks(self) -> None:
+        agent = _FakeAgent(reply="hello", chunks=["he", "llo"])
+
+        def transform(update: AgentResponseUpdate) -> AgentResponseUpdate:
+            return AgentResponseUpdate(contents=[Content.from_text(update.text.upper())], role="assistant")
+
+        host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel(stream_update_hook=transform)])
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "hi", "stream": True})
+
+        assert r.status_code == 200
+        assert '"delta":"HE"' in r.text
+        assert '"delta":"LLO"' in r.text
+        # Stream update hooks are update-only; they do not rewrite get_final_response().
+        assert '"text":"hello"' in r.text
+
+    def test_sse_completed_preserves_streamed_multimodal_updates_when_finalize_fails(self) -> None:
+        class _MultimodalStream:
+            def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
+                async def _gen() -> AsyncIterator[AgentResponseUpdate]:
+                    yield AgentResponseUpdate(
+                        contents=[
+                            Content.from_text("caption"),
+                            Content.from_text_reasoning(id="rs_1", text="thinking"),
+                            Content.from_function_call("call_1", "lookup", arguments={"city": "Seattle"}),
+                            Content.from_uri("https://example.com/cat.png", media_type="image/png"),
+                        ],
+                        role="assistant",
+                    )
+
+                return _gen()
+
+            async def get_final_response(self) -> _FakeAgentResponse:
+                raise RuntimeError("finalize unavailable")
+
+        class _MultimodalAgent(_FakeAgent):
+            def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
+                self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
+                if stream:
+                    return _MultimodalStream()
+                raise AssertionError("non-streaming path not exercised here")
+
+        host = AgentFrameworkHost(target=_MultimodalAgent(), channels=[ResponsesChannel()])
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "hi", "stream": True})
+
+        assert r.status_code == 200
+        assert "event: response.output_item.added" in r.text
+        assert "event: response.output_item.done" in r.text
+        events = [line[len("event: ") :] for line in r.text.splitlines() if line.startswith("event: ")]
+        assert "response.content_part.added" in events
+        assert "response.output_text.done" in events
+        assert "response.reasoning_text.delta" in events
+        assert "response.reasoning_text.done" in events
+        assert "response.function_call_arguments.delta" in events
+        assert "response.function_call_arguments.done" in events
+        content_part_added = _sse_payload(r.text, "response.content_part.added")
+        assert content_part_added["part"] == {"annotations": [], "text": "", "type": "output_text"}
+        added_items = [
+            json.loads(line[len("data: ") :])["item"]
+            for line in r.text.splitlines()
+            if line.startswith("data: ") and '"type":"response.output_item.added"' in line
+        ]
+        assert [item["type"] for item in added_items] == [
+            "message",
+            "reasoning",
+            "function_call",
+            "function_call_output",
+        ]
+        assert added_items[0]["content"] == []
+        assert added_items[1]["content"] == []
+        assert added_items[2]["name"] == "lookup"
+        assert added_items[2]["arguments"] == ""
+        assert added_items[3]["output"] == [
+            {"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
+        ]
+        completed = _sse_payload(r.text, "response.completed")
+        assert completed["response"]["output"][0]["content"][0]["text"] == "caption"
+        assert completed["response"]["output"][1]["content"][0]["text"] == "thinking"
+        assert completed["response"]["output"][2]["name"] == "lookup"
+        assert completed["response"]["output"][3]["output"] == [
+            {"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
+        ]
+
+    def test_sse_emits_failed_when_stream_raises(self) -> None:
+        # Regression: ResponseOutputMessage.status only accepts in_progress/
+        # completed/incomplete, so building an OpenAIResponse with status="failed"
+        # used to crash with a pydantic ValidationError. The channel must map the
+        # nested message status to "incomplete" while keeping the top-level
+        # Response.status="failed".
+        class _BoomStream:
+            def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
+                async def _gen() -> AsyncIterator[AgentResponseUpdate]:
+                    yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant")
+                    raise RuntimeError("upstream blew up")
+
+                return _gen()
+
+            async def get_final_response(self) -> _FakeAgentResponse:  # pragma: no cover
+                return _FakeAgentResponse(text="")
+
+        class _BoomAgent(_FakeAgent):
+            def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
+                self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
+                if stream:
+                    return _BoomStream()
+                raise AssertionError("non-streaming path not exercised here")
+
+        host = AgentFrameworkHost(target=_BoomAgent(), channels=[ResponsesChannel()])
+        with TestClient(host.app) as client:
+            r = client.post("/responses", json={"input": "hi", "stream": True})
+            assert r.status_code == 200
+            body = r.text
+
+        events = [line[len("event: ") :] for line in body.splitlines() if line.startswith("event: ")]
+        assert events[0] == "response.created"
+        assert events[-1] == "response.failed"
+        # The failed envelope must serialize cleanly — i.e. no ValidationError raised.
+        assert "upstream blew up" in body

python/packages/hosting-responses/tests/hosting_responses/test_parsing.py

@@ -0,0 +1,169 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the OpenAI Responses request-body parser."""
+
+from __future__ import annotations
+
+import pytest
+
+from agent_framework_hosting_responses import (
+    messages_from_responses_input,
+    parse_responses_identity,
+    parse_responses_request,
+)
+
+
+class TestMessagesFromResponsesInput:
+    def test_string_input_becomes_single_user_message(self) -> None:
+        msgs = messages_from_responses_input("hello")
+        assert len(msgs) == 1
+        assert msgs[0].role == "user"
+        assert msgs[0].text == "hello"
+
+    def test_input_text_items_collapse_into_one_user_message(self) -> None:
+        msgs = messages_from_responses_input([{"type": "input_text", "text": "a"}, {"type": "input_text", "text": "b"}])
+        assert len(msgs) == 1
+        assert msgs[0].role == "user"
+        assert msgs[0].text == "a b"
+
+    def test_message_envelope_with_string_content(self) -> None:
+        msgs = messages_from_responses_input([
+            {"type": "message", "role": "system", "content": "be brief"},
+            {"type": "message", "role": "user", "content": "hi"},
+        ])
+        assert [m.role for m in msgs] == ["system", "user"]
+        assert msgs[0].text == "be brief"
+
+    def test_message_envelope_with_content_parts(self) -> None:
+        msgs = messages_from_responses_input([
+            {
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": "describe this"}],
+            }
+        ])
+        assert msgs[0].text == "describe this"
+
+    def test_message_envelope_rejects_non_object_content_item(self) -> None:
+        with pytest.raises(ValueError, match="content.*object"):
+            messages_from_responses_input([{"type": "message", "role": "user", "content": ["bad"]}])
+
+    def test_message_envelope_rejects_invalid_content_shape(self) -> None:
+        with pytest.raises(ValueError, match="content.*string or list"):
+            messages_from_responses_input([{"type": "message", "role": "user", "content": 42}])
+
+    def test_input_file_via_url(self) -> None:
+        msgs = messages_from_responses_input([
+            {"type": "input_file", "file_url": "https://example.com/report.pdf", "mime_type": "application/pdf"}
+        ])
+        assert msgs[0].contents[0].uri == "https://example.com/report.pdf"
+
+    def test_input_file_via_file_id(self) -> None:
+        msgs = messages_from_responses_input([{"type": "input_file", "file_id": "file_123"}])
+        assert msgs[0].contents[0].file_id == "file_123"
+
+    def test_input_file_missing_anchor_raises(self) -> None:
+        with pytest.raises(ValueError, match="input_file"):
+            messages_from_responses_input([{"type": "input_file"}])
+
+    def test_pending_text_flushes_before_message_envelope(self) -> None:
+        msgs = messages_from_responses_input([
+            {"type": "input_text", "text": "first"},
+            {"type": "message", "role": "user", "content": "second"},
+        ])
+        assert len(msgs) == 2
+        assert msgs[0].text == "first"
+        assert msgs[1].text == "second"
+
+    def test_image_url_via_string(self) -> None:
+        msgs = messages_from_responses_input([{"type": "input_image", "image_url": "https://example.com/cat.png"}])
+        assert len(msgs) == 1
+        # Image content present.
+        assert any(getattr(c, "uri", None) == "https://example.com/cat.png" for c in msgs[0].contents)
+
+    def test_image_url_via_object(self) -> None:
+        msgs = messages_from_responses_input([
+            {"type": "input_image", "image_url": {"url": "https://example.com/cat.png"}}
+        ])
+        assert any(getattr(c, "uri", None) == "https://example.com/cat.png" for c in msgs[0].contents)
+
+    def test_unknown_input_type_raises(self) -> None:
+        with pytest.raises(ValueError, match="Unsupported"):
+            messages_from_responses_input([{"type": "weird"}])
+
+    def test_empty_list_raises(self) -> None:
+        with pytest.raises(ValueError, match="non-empty"):
+            messages_from_responses_input([])
+
+    def test_non_string_non_list_raises(self) -> None:
+        with pytest.raises(ValueError):
+            messages_from_responses_input(42)  # type: ignore[arg-type]
+
+    def test_image_url_missing_raises(self) -> None:
+        with pytest.raises(ValueError, match="image_url"):
+            messages_from_responses_input([{"type": "input_image"}])
+
+
+class TestParseResponsesRequest:
+    def test_known_fields_remapped_and_unknown_forwarded(self) -> None:
+        _, opts, _ = parse_responses_request({
+            "input": "hi",
+            "instructions": "be brief",
+            "temperature": 0.4,
+            "top_p": 0.9,
+            "tool_choice": "auto",
+            "max_output_tokens": 256,
+            "parallel_tool_calls": False,
+            "truncation": "auto",
+            "reasoning": {"effort": "low"},
+        })
+        # Known remaps applied.
+        assert opts["max_tokens"] == 256
+        assert opts["allow_multiple_tool_calls"] is False
+        # Straight-through fields present.
+        assert opts["temperature"] == 0.4
+        assert opts["instructions"] == "be brief"
+        assert opts["truncation"] == "auto"
+        # Transport/session keys excluded.
+        for key in ("input", "stream", "previous_response_id"):
+            assert key not in opts
+
+    def test_model_passes_through_transport_keys_excluded(self) -> None:
+        _, opts, _ = parse_responses_request({
+            "input": "x",
+            "model": "gpt-x",
+            "stream": True,
+            "previous_response_id": "r",
+        })
+        for key in ("input", "stream", "previous_response_id"):
+            assert key not in opts
+        # model passes through — not a transport key; run_hook decides what to do with it.
+        assert opts["model"] == "gpt-x"
+
+    def test_none_values_dropped(self) -> None:
+        _, opts, _ = parse_responses_request({"input": "x", "temperature": None})
+        assert "temperature" not in opts
+
+    def test_previous_response_id_becomes_session(self) -> None:
+        _, _, sess = parse_responses_request({"input": "x", "previous_response_id": "resp_42"})
+        assert sess is not None
+        assert sess.isolation_key == "resp_42"
+
+
+class TestParseResponsesIdentity:
+    def test_safety_identifier_preferred(self) -> None:
+        ident = parse_responses_identity({"safety_identifier": "abc", "user": "legacy"}, "responses")
+        assert ident is not None
+        assert ident.native_id == "abc"
+        assert ident.channel == "responses"
+
+    def test_fallback_to_user(self) -> None:
+        ident = parse_responses_identity({"user": "legacy"}, "responses")
+        assert ident is not None
+        assert ident.native_id == "legacy"
+
+    def test_returns_none_when_absent(self) -> None:
+        assert parse_responses_identity({}, "responses") is None
+
+    def test_returns_none_for_non_string(self) -> None:
+        assert parse_responses_identity({"safety_identifier": 42}, "responses") is None

python/packages/hosting/LICENSE

@@ -0,0 +1,21 @@
+    MIT License
+
+    Copyright (c) Microsoft Corporation.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"), to deal
+    in the Software without restriction, including without limitation the rights
+    to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+    copies of the Software, and to permit persons to whom the Software is
+    furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all
+    copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+    AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+    OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+    SOFTWARE

python/packages/hosting/README.md

@@ -0,0 +1,122 @@
+# agent-framework-hosting
+
+Multi-channel hosting for Microsoft Agent Framework agents.
+
+`agent-framework-hosting` lets you serve a single agent or workflow target
+through one or more **channels**. The host owns one Starlette ASGI app,
+route/lifecycle composition, and per-`isolation_key` session resolution.
+Each channel owns its protocol parsing and response rendering.
+
+The base package contains only channel-neutral plumbing:
+
+- `AgentFrameworkHost` — the Starlette host.
+- `Channel` — the channel protocol.
+- `ChannelRequest` / `ChannelSession` / `ChannelIdentity` — the request
+  envelope and optional channel metadata.
+- `ChannelContext` / `ChannelContribution` / `ChannelCommand` — channel-side
+  hooks for invoking the target and contributing routes, commands, and
+  lifecycle callbacks.
+- `ChannelRunHook` / `ChannelResponseHook` / `ChannelStreamUpdateHook` —
+  host-invoked customization seams.
+
+`ChannelStreamUpdateHook` applies to streamed updates only. It is not a
+substitute for final-response redaction.
+
+Concrete channels live in their own packages so you only install what you use:
+
+| Package | Transport |
+|---|---|
+| `agent-framework-hosting-responses` | OpenAI Responses API |
+
+Additional channel packages can build on the same host contract without adding
+their protocol dependencies to the base package.
+
+## Install
+
+```bash
+pip install agent-framework-hosting agent-framework-hosting-responses
+# or with Hypercorn pre-installed for the demo `host.serve(...)` helper
+pip install "agent-framework-hosting[serve]" agent-framework-hosting-responses
+# add the [disk] extra to persist reset-session aliases
+pip install "agent-framework-hosting[disk]"
+```
+
+## Quickstart
+
+```python
+from agent_framework.openai import OpenAIChatClient
+from agent_framework_hosting import AgentFrameworkHost, Channel
+
+agent = OpenAIChatClient().as_agent(name="Assistant")
+
+# Add channels from sibling packages, e.g. `agent-framework-hosting-responses`
+# exposes a `ResponsesChannel` that serves the OpenAI Responses API.
+channels: list[Channel] = []
+
+host = AgentFrameworkHost(target=agent, channels=channels)
+host.serve(port=8000)
+```
+
+## Session state and workflow checkpoints
+
+By default the host keeps live `AgentSession` objects and reset-session aliases
+in memory. Channels opt into continuity by setting
+`ChannelRequest.session = ChannelSession(isolation_key=...)`; requests with the
+same isolation key reuse the same host-created session.
+
+The host treats `isolation_key` as an opaque partition key. Each channel or
+hosting environment decides where that key comes from:
+
+- protocol headers supplied by a trusted platform,
+- request body fields such as a previous response or conversation ID,
+- route/path parameters,
+- channel-native metadata such as chat/user IDs, or
+- environment-provided context in an ephemeral host.
+
+The host should be able to carry any of those sources as long as the channel or
+platform has already authenticated and authorized the caller before passing the
+key to `ChannelSession`.
+
+The built-in request-context helper recognizes the `x-agent-user-isolation-key`
+and `x-agent-chat-isolation-key` header names because some hosting
+environments, including Foundry Hosted Agents, already use them. Reusing those
+header names does **not** mean `agent-framework-hosting` is the supported way to
+run on Foundry Hosted Agents; use `agent-framework-foundry-hosting` for that
+hosting surface.
+
+For long-running deployments that need `reset_session(...)` aliases to survive
+restart, pass `state_dir`:
+
+```python
+host = AgentFrameworkHost(
+    target=agent,
+    channels=channels,
+    state_dir="./.host-state",
+)
+```
+
+This creates `./.host-state/sessions/` and stores only lightweight alias
+bookkeeping. Live `AgentSession` objects are still rehydrated lazily by the
+configured history provider on the next turn.
+
+For workflow targets, `checkpoint_location=...` is the clearest way to enable
+checkpoint persistence. As a convenience, `state_dir="./.host-state"` also
+derives `./.host-state/checkpoints/` for workflow targets. Use the mapping form
+when you want only one component:
+
+```python
+from agent_framework_hosting import HostStatePaths
+
+host = AgentFrameworkHost(
+    target=workflow,
+    channels=channels,
+    state_dir=HostStatePaths(
+        sessions="/var/lib/myapp/sessions",
+        checkpoints="/var/lib/myapp/checkpoints",
+    ),
+)
+```
+
+Cross-channel identity linking, multicast delivery, background runs,
+continuation tokens, and durable delivery runners are follow-up enhancements,
+not part of this v1 host contract.

python/packages/hosting/agent_framework_hosting/__init__.py

@@ -0,0 +1,66 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Multi-channel hosting for Microsoft Agent Framework agents.
+
+Serve a single agent target through one or more **channels** — pluggable
+adapters that expose the target over different transports. The base
+package contains only the channel-neutral plumbing; concrete channels
+ship in their own packages, such as ``agent-framework-hosting-responses``,
+so users install only what they need.
+"""
+
+import importlib.metadata
+
+from ._host import AgentFrameworkHost, ChannelContext, logger
+from ._isolation import (
+    ISOLATION_HEADER_CHAT,
+    ISOLATION_HEADER_USER,
+    IsolationKeys,
+    get_current_isolation_keys,
+    reset_current_isolation_keys,
+    set_current_isolation_keys,
+)
+from ._types import (
+    Channel,
+    ChannelCommand,
+    ChannelCommandContext,
+    ChannelContribution,
+    ChannelIdentity,
+    ChannelRequest,
+    ChannelResponseHook,
+    ChannelRunHook,
+    ChannelSession,
+    ChannelStreamUpdateHook,
+    HostedRunResult,
+    HostStatePaths,
+)
+
+try:
+    __version__ = importlib.metadata.version(__name__)
+except importlib.metadata.PackageNotFoundError:
+    __version__ = "0.0.0"
+
+__all__ = [
+    "ISOLATION_HEADER_CHAT",
+    "ISOLATION_HEADER_USER",
+    "AgentFrameworkHost",
+    "Channel",
+    "ChannelCommand",
+    "ChannelCommandContext",
+    "ChannelContext",
+    "ChannelContribution",
+    "ChannelIdentity",
+    "ChannelRequest",
+    "ChannelResponseHook",
+    "ChannelRunHook",
+    "ChannelSession",
+    "ChannelStreamUpdateHook",
+    "HostStatePaths",
+    "HostedRunResult",
+    "IsolationKeys",
+    "__version__",
+    "get_current_isolation_keys",
+    "logger",
+    "reset_current_isolation_keys",
+    "set_current_isolation_keys",
+]

python/packages/hosting/agent_framework_hosting/_isolation.py

@@ -0,0 +1,82 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Per-request isolation keys for host/platform-provided request context.
+
+``ChannelSession.isolation_key`` is the host's generic session partition key,
+but different channels and platforms discover that key from different places:
+protocol headers, request bodies, URL/path segments, webhook metadata, or
+environment-provided context for ephemeral hosts.
+
+This module covers the request-context case where a platform provides
+isolation outside the channel payload. The Foundry Hosted Agents runtime, for
+example, injects two well-known headers on requests it forwards to the user's
+container:
+
+* ``x-agent-user-isolation-key`` — opaque per-user partition key
+* ``x-agent-chat-isolation-key`` — opaque per-conversation partition key
+
+The generic host intentionally reuses those header names so the same isolation
+context can be consumed by supporting providers. Reusing the names does **not**
+mean ``agent-framework-hosting`` is a supported way to run on Foundry Hosted
+Agents; use ``agent-framework-foundry-hosting`` for that hosting surface.
+
+When those headers are present the host-installed ASGI middleware pushes them
+into :data:`current_isolation_keys` for the duration of the request, then
+resets it. Channels may still choose a different session key source and pass it
+directly via ``ChannelSession(isolation_key=...)``.
+
+The contextvar holds a plain :class:`IsolationKeys` mapping; conversion to
+provider-specific types happens at the consuming provider so this module has no
+provider dependencies.
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar, Token
+
+__all__ = [
+    "ISOLATION_HEADER_CHAT",
+    "ISOLATION_HEADER_USER",
+    "IsolationKeys",
+    "current_isolation_keys",
+    "get_current_isolation_keys",
+    "reset_current_isolation_keys",
+    "set_current_isolation_keys",
+]
+
+
+ISOLATION_HEADER_USER = "x-agent-user-isolation-key"
+ISOLATION_HEADER_CHAT = "x-agent-chat-isolation-key"
+
+
+class IsolationKeys:
+    """Per-request isolation keys lifted from host/platform context."""
+
+    def __init__(self, user_key: str | None = None, chat_key: str | None = None) -> None:
+        self.user_key = user_key
+        self.chat_key = chat_key
+
+    @property
+    def is_empty(self) -> bool:
+        return self.user_key is None and self.chat_key is None
+
+
+current_isolation_keys: ContextVar[IsolationKeys | None] = ContextVar(
+    "agent_framework_hosting_isolation_keys",
+    default=None,
+)
+
+
+def get_current_isolation_keys() -> IsolationKeys | None:
+    """Return the isolation keys bound to the current request, if any."""
+    return current_isolation_keys.get()
+
+
+def set_current_isolation_keys(keys: IsolationKeys | None) -> Token[IsolationKeys | None]:
+    """Bind ``keys`` to the current async context and return a reset token."""
+    return current_isolation_keys.set(keys)
+
+
+def reset_current_isolation_keys(token: Token[IsolationKeys | None]) -> None:
+    """Restore the isolation contextvar to its prior value."""
+    current_isolation_keys.reset(token)

python/packages/hosting/agent_framework_hosting/_persistence.py

@@ -0,0 +1,128 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Shared persistence primitives for the hosting package.
+
+The simplified hosting core keeps disk persistence only for session aliases
+created by :meth:`AgentFrameworkHost.reset_session` and for workflow
+checkpoint path derivation. The on-disk session-alias store uses the optional
+``diskcache`` package installed via the ``[disk]`` extra.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import importlib
+import os
+import sys
+from collections.abc import Mapping
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from ._types import HostStatePaths
+
+_KNOWN_COMPONENTS: tuple[str, ...] = ("sessions", "checkpoints")
+
+
+def load_diskcache() -> Any:
+    """Lazy-import :mod:`diskcache` with a helpful error when missing."""
+    try:
+        return importlib.import_module("diskcache")
+    except ImportError as exc:  # pragma: no cover - exercised via tests by monkeypatching
+        raise ImportError(
+            "agent-framework-hosting was asked to persist session aliases to disk "
+            "(state_dir['sessions'] is set) but the optional `diskcache` dependency "
+            "is not installed. Install the disk extra: "
+            "`pip install 'agent-framework-hosting[disk]`."
+        ) from exc
+
+
+def acquire_state_dir_lock(component_dir: Path) -> Any:
+    """Acquire an exclusive single-owner lock on a component's state dir.
+
+    Raises:
+        RuntimeError: If another process already holds the lock.
+    """
+    component_dir.mkdir(parents=True, exist_ok=True)
+    lock_path = component_dir / ".lock"
+    fh = open(lock_path, "a+", encoding="utf-8")  # noqa: SIM115 - kept open for lifetime
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            try:
+                msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
+            except OSError as exc:
+                fh.close()
+                raise RuntimeError(
+                    f"Another process already holds the hosting state lock at {lock_path}. "
+                    "Point each host at its own state_dir."
+                ) from exc
+        else:
+            import fcntl
+
+            try:
+                fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except OSError as exc:
+                fh.close()
+                raise RuntimeError(
+                    f"Another process already holds the hosting state lock at {lock_path}. "
+                    "Point each host at its own state_dir."
+                ) from exc
+    except RuntimeError:
+        raise
+    except Exception:
+        fh.close()
+        raise
+    return fh
+
+
+def release_state_dir_lock(handle: Any) -> None:
+    """Release a lock previously acquired by :func:`acquire_state_dir_lock`."""
+    if handle is None:
+        return
+    with contextlib.suppress(Exception):
+        handle.close()
+
+
+def normalize_state_dir(
+    state_dir: str | os.PathLike[str] | HostStatePaths | Mapping[str, str | os.PathLike[str]] | None,
+) -> dict[str, Path | None]:
+    """Resolve the host-level ``state_dir`` parameter into a per-component map.
+
+    Accepts ``None``, a single root path, or a mapping with ``sessions`` and
+    ``checkpoints`` keys. Unknown keys raise ``ValueError`` so obsolete
+    ``runner`` / ``links`` configuration is rejected instead of silently
+    doing nothing.
+    """
+    result: dict[str, Path | None] = {name: None for name in _KNOWN_COMPONENTS}
+    if state_dir is None:
+        return result
+
+    if isinstance(state_dir, (str, os.PathLike)):
+        root = Path(os.fspath(state_dir))
+        for name in _KNOWN_COMPONENTS:
+            result[name] = root / name
+        return result
+
+    if isinstance(state_dir, Mapping):
+        unknown = [k for k in state_dir if k not in _KNOWN_COMPONENTS]
+        if unknown:
+            raise ValueError(
+                f"state_dir mapping contains unknown component key(s): {unknown!r}. "
+                f"Known components are: {list(_KNOWN_COMPONENTS)!r}."
+            )
+        for name in _KNOWN_COMPONENTS:
+            raw_value: Any = state_dir.get(name)
+            if raw_value is None:
+                result[name] = None
+                continue
+            if isinstance(raw_value, (str, os.PathLike)):
+                result[name] = Path(os.fspath(raw_value))
+            else:
+                raise TypeError(f"state_dir[{name!r}] must be a str or PathLike — got {type(raw_value).__name__}")
+        return result
+
+    raise TypeError(
+        f"state_dir must be a str, PathLike, HostStatePaths mapping, or None — got {type(state_dir).__name__}"
+    )

python/packages/hosting/agent_framework_hosting/_state_store.py

@@ -0,0 +1,146 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Disk-backed wrapper for the host's session-alias map.
+
+``AgentFrameworkHost.reset_session(isolation_key)`` rotates future requests for
+that isolation key onto a new session id. Persisting the alias map lets that
+rotation survive a host restart without introducing cross-channel identity or
+delivery state into the core host.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, TypeVar
+
+from ._persistence import (
+    acquire_state_dir_lock,
+    load_diskcache,
+    release_state_dir_lock,
+)
+
+logger = logging.getLogger(__name__)
+
+_V = TypeVar("_V")
+_ALIASES_PREFIX = "aliases:"
+
+
+class SessionsStateStore:
+    """One disk cache + lock for host-side session aliases."""
+
+    def __init__(self, sessions_dir: str | os.PathLike[str]) -> None:
+        self._sessions_dir: Path = Path(os.fspath(sessions_dir))
+        diskcache = load_diskcache()
+        self._lock_handle: Any = acquire_state_dir_lock(self._sessions_dir)
+        try:
+            self._cache: Any = diskcache.Cache(str(self._sessions_dir))
+        except Exception:
+            release_state_dir_lock(self._lock_handle)
+            self._lock_handle = None
+            raise
+
+    @property
+    def cache(self) -> Any:
+        """Return the underlying :mod:`diskcache` Cache."""
+        return self._cache
+
+    def close(self) -> None:
+        """Close the cache and release the directory lock."""
+        if self._cache is not None:
+            try:
+                self._cache.close()
+            except Exception:  # pragma: no cover - close errors aren't actionable
+                logger.exception("SessionsStateStore: failed to close cache cleanly")
+            self._cache = None
+        if self._lock_handle is not None:
+            release_state_dir_lock(self._lock_handle)
+            self._lock_handle = None
+
+
+class _PersistedDict(dict[str, _V]):
+    """Drop-in :class:`dict` whose mutations mirror to a diskcache prefix."""
+
+    def __init__(
+        self,
+        store: SessionsStateStore,
+        key_prefix: str,
+        initial: Mapping[str, _V] | None = None,
+    ) -> None:
+        super().__init__()
+        self._store = store
+        self._prefix = key_prefix
+        cache: Any = store.cache
+        for raw_key in cache.iterkeys():
+            if not isinstance(raw_key, str) or not raw_key.startswith(key_prefix):
+                continue
+            try:
+                value: Any = cache.get(raw_key)
+            except Exception:
+                logger.exception("SessionsStateStore: failed to rehydrate %s; skipping", raw_key)
+                continue
+            logical_key = raw_key[len(key_prefix) :]
+            super().__setitem__(logical_key, value)
+        if initial:
+            for key, value in initial.items():
+                self[key] = value
+
+    def __setitem__(self, key: str, value: _V) -> None:
+        super().__setitem__(key, value)
+        try:
+            self._store.cache.set(self._prefix + key, value)
+        except Exception:  # pragma: no cover - cache write failures aren't actionable
+            logger.exception("SessionsStateStore: failed to persist %s%s", self._prefix, key)
+
+    def __delitem__(self, key: str) -> None:
+        super().__delitem__(key)
+        try:
+            del self._store.cache[self._prefix + key]
+        except KeyError:
+            pass
+        except Exception:  # pragma: no cover - cache write failures aren't actionable
+            logger.exception("SessionsStateStore: failed to evict %s%s", self._prefix, key)
+
+    def pop(self, key: str, *args: Any) -> _V:
+        """Mirror ``dict.pop`` to disk."""
+        value: _V = super().pop(key, *args)
+        try:
+            del self._store.cache[self._prefix + key]
+        except KeyError:
+            pass
+        except Exception:  # pragma: no cover
+            logger.exception("SessionsStateStore: failed to evict %s%s", self._prefix, key)
+        return value
+
+    def clear(self) -> None:
+        """Mirror ``dict.clear`` to disk."""
+        keys = list(self.keys())
+        super().clear()
+        cache = self._store.cache
+        for key in keys:
+            try:
+                del cache[self._prefix + key]
+            except KeyError:
+                pass
+            except Exception:  # pragma: no cover
+                logger.exception("SessionsStateStore: failed to evict %s%s during clear", self._prefix, key)
+
+    def update(  # type: ignore[override]
+        self,
+        other: Mapping[str, _V] | None = None,
+        /,
+        **kwargs: _V,
+    ) -> None:
+        """Mirror ``dict.update`` to disk one item at a time."""
+        if other is not None:
+            for key in other:
+                self[key] = other[key]
+        for key, value in kwargs.items():
+            self[key] = value
+
+
+def build_session_aliases(store: SessionsStateStore) -> dict[str, str]:
+    """Return the disk-backed session-alias map for ``store``."""
+    return _PersistedDict[str](store, _ALIASES_PREFIX)

python/packages/hosting/agent_framework_hosting/_types.py

@@ -0,0 +1,212 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+# ``ChannelRequest`` is the only intentional dataclass here (callers use
+# ``dataclasses.replace`` on it in run hooks). The other types are plain
+# Python classes by preference, so the "could be a dataclass" lint is muted
+# at the file level.
+# ruff: noqa: B903
+
+"""Channel-neutral request envelope and channel protocol types.
+
+These types form the boundary between the host and individual channels.
+A channel parses its native payload, builds a :class:`ChannelRequest`, and
+hands it to :class:`ChannelContext.run` (or ``run_stream``) on the host.
+The channel owns rendering the result back onto its originating protocol.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Awaitable, Callable, Mapping, Sequence
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Generic, Protocol, TypedDict, TypeVar, runtime_checkable
+
+from agent_framework import (
+    AgentResponseUpdate,
+    AgentRunInputs,
+)
+from starlette.routing import BaseRoute
+
+if TYPE_CHECKING:
+    from ._host import ChannelContext
+
+
+class ChannelSession:
+    """Channel-supplied session hint.
+
+    The host turns this into an ``AgentSession`` keyed by ``isolation_key`` so
+    every distinct end user gets their own context-provider state (e.g. one
+    ``FileHistoryProvider`` JSONL file per user).
+    """
+
+    def __init__(self, isolation_key: str | None = None) -> None:
+        self.isolation_key = isolation_key
+
+
+class ChannelIdentity:
+    """Channel-native identity metadata observed on a request.
+
+    The simplified hosting core records this only on the persisted input
+    message's ``additional_properties["hosting"]`` block and forwards it
+    through run/response hooks. Cross-channel linking and recipient lookup are
+    follow-up concerns, not part of the v1 host contract.
+    """
+
+    def __init__(
+        self,
+        channel: str,
+        native_id: str,
+        attributes: Mapping[str, Any] | None = None,
+    ) -> None:
+        self.channel = channel
+        self.native_id = native_id
+        self.attributes: Mapping[str, Any] = attributes if attributes is not None else dict()
+
+
+@dataclass
+class ChannelRequest:
+    """Uniform invocation envelope every channel produces from its native payload.
+
+    Kept as a dataclass so app authors can use ``dataclasses.replace(...)`` in
+    run hooks to produce a modified envelope without re-listing every field.
+    """
+
+    channel: str
+    operation: str
+    input: AgentRunInputs
+    session: ChannelSession | None = None
+    options: Mapping[str, Any] | None = None
+    session_mode: str = "auto"
+    metadata: Mapping[str, Any] = field(default_factory=lambda: {})
+    attributes: Mapping[str, Any] = field(default_factory=lambda: {})
+    stream: bool = False
+    identity: ChannelIdentity | None = None
+
+
+class ChannelCommand:
+    """A discoverable command a channel exposes to its users (e.g. ``/reset``)."""
+
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        handle: Callable[[ChannelCommandContext], Awaitable[None]],
+    ) -> None:
+        self.name = name
+        self.description = description
+        self.handle = handle
+
+
+class ChannelCommandContext:
+    """Context passed to a :class:`ChannelCommand` handler."""
+
+    def __init__(
+        self,
+        request: ChannelRequest,
+        reply: Callable[[str], Awaitable[None]],
+    ) -> None:
+        self.request = request
+        self.reply = reply
+
+
+_EMPTY_ROUTES: tuple[BaseRoute, ...] = ()
+_EMPTY_COMMANDS: tuple[ChannelCommand, ...] = ()
+_EMPTY_LIFECYCLE: tuple[Callable[[], Awaitable[None]], ...] = ()
+
+
+class ChannelContribution:
+    """Routes, commands, and lifecycle hooks a channel contributes to the host."""
+
+    def __init__(
+        self,
+        routes: Sequence[BaseRoute] = _EMPTY_ROUTES,
+        commands: Sequence[ChannelCommand] = _EMPTY_COMMANDS,
+        on_startup: Sequence[Callable[[], Awaitable[None]]] = _EMPTY_LIFECYCLE,
+        on_shutdown: Sequence[Callable[[], Awaitable[None]]] = _EMPTY_LIFECYCLE,
+    ) -> None:
+        self.routes = routes
+        self.commands = commands
+        self.on_startup = on_startup
+        self.on_shutdown = on_shutdown
+
+
+class _Unset:
+    """Sentinel for ``HostedRunResult.replace`` overrides.
+
+    Distinguishes "caller did not pass this kwarg" from "caller passed
+    ``None`` explicitly" — needed because ``session`` is ``None`` in
+    many envelopes and we want the no-arg call to preserve it.
+    """
+
+
+_UNSET = _Unset()
+
+
+TResult = TypeVar("TResult")
+
+
+class HostedRunResult(Generic[TResult]):
+    """Channel-neutral envelope around the target's full-fidelity result.
+
+    The host does not flatten or pre-shape the target output. Channels and
+    response hooks read the underlying result type directly and serialize the
+    subset their wire format can carry.
+    """
+
+    def __init__(
+        self,
+        result: TResult,
+        *,
+        session: Any | None = None,
+    ) -> None:
+        self.result = result
+        self.session = session
+
+    def replace(
+        self,
+        *,
+        result: TResult | _Unset = _UNSET,
+        session: Any | _Unset | None = _UNSET,
+    ) -> HostedRunResult[TResult]:
+        """Return a shallow copy with the supplied fields overridden."""
+        new: HostedRunResult[TResult] = HostedRunResult.__new__(HostedRunResult)  # pyright: ignore[reportUnknownVariableType]
+        new.result = self.result if isinstance(result, _Unset) else result
+        new.session = self.session if isinstance(session, _Unset) else session
+        return new
+
+
+class HostStatePaths(TypedDict, total=False):
+    """Per-component disk paths for host-managed state.
+
+    Only session aliases and workflow checkpoints remain in the simplified
+    host. Linking stores, active-channel maps, identity registries, and runner
+    queues are follow-up concerns.
+    """
+
+    sessions: str | os.PathLike[str]
+    """Where the host persists session aliases created by ``reset_session``."""
+
+    checkpoints: str | os.PathLike[str]
+    """Where the host persists workflow checkpoints for ``Workflow`` targets."""
+
+
+ChannelStreamUpdateHook = Callable[
+    [AgentResponseUpdate],
+    "AgentResponseUpdate | Awaitable[AgentResponseUpdate | None] | None",
+]
+
+
+ChannelRunHook = Callable[..., "Awaitable[ChannelRequest] | ChannelRequest"]
+
+
+ChannelResponseHook = Callable[..., "Awaitable[HostedRunResult[Any]] | HostedRunResult[Any]"]
+
+
+@runtime_checkable
+class Channel(Protocol):
+    """A pluggable adapter that exposes one transport on the host."""
+
+    name: str
+    path: str
+
+    def contribute(self, context: ChannelContext) -> ChannelContribution: ...

python/packages/hosting/pyproject.toml

@@ -0,0 +1,92 @@
+[project]
+name = "agent-framework-hosting"
+description = "Multi-channel hosting for Microsoft Agent Framework agents."
+authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
+readme = "README.md"
+requires-python = ">=3.10"
+version = "1.0.0a260424"
+license-files = ["LICENSE"]
+urls.homepage = "https://aka.ms/agent-framework"
+urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
+urls.issues = "https://github.com/microsoft/agent-framework/issues"
+classifiers = [
+  "License :: OSI Approved :: MIT License",
+  "Development Status :: 3 - Alpha",
+  "Intended Audience :: Developers",
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
+  "Typing :: Typed",
+]
+dependencies = [
+    "agent-framework-core>=1.2.0,<2",
+    "starlette>=0.37",
+]
+
+[project.optional-dependencies]
+serve = [
+    "hypercorn>=0.17",
+]
+disk = [
+    "diskcache>=5.6",
+]
+
+[tool.uv]
+prerelease = "if-necessary-or-explicit"
+environments = [
+    "sys_platform == 'darwin'",
+    "sys_platform == 'linux'",
+    "sys_platform == 'win32'"
+]
+
+[tool.uv-dynamic-versioning]
+fallback-version = "0.0.0"
+
+[tool.pytest.ini_options]
+testpaths = 'tests'
+addopts = "-ra -q -r fEX"
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+filterwarnings = []
+timeout = 120
+markers = [
+    "integration: marks tests as integration tests that require external services",
+]
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+
+[tool.coverage.run]
+omit = [
+    "**/__init__.py"
+]
+
+[tool.pyright]
+extends = "../../pyproject.toml"
+include = ["agent_framework_hosting"]
+exclude = ['tests']
+
+[tool.bandit]
+targets = ["agent_framework_hosting"]
+exclude_dirs = ["tests"]
+
+[tool.poe]
+executor.type = "uv"
+include = "../../shared_tasks.toml"
+
+[tool.poe.tasks.test]
+help = "Run the default unit test suite for this package."
+cmd = 'pytest -m "not integration" --cov=agent_framework_hosting --cov-report=term-missing:skip-covered tests'
+
+[build-system]
+requires = ["flit-core >= 3.11,<4.0"]
+build-backend = "flit_core.buildapi"
+
+[dependency-groups]
+dev = [
+    "httpx>=0.28.1",
+]

python/packages/hosting/tests/hosting/_workflow_fixtures.py

@@ -0,0 +1,45 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Workflow fixtures for hosting tests.
+
+Defined in a module that does not use ``from __future__ import annotations``
+because the workflow handler validation reflects on real annotation objects
+rather than stringified forms.
+"""
+
+from typing import Any
+
+from agent_framework import Executor, Workflow, WorkflowBuilder, WorkflowContext, handler
+
+
+class _UpperExecutor(Executor):
+    @handler
+    async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
+        await ctx.yield_output(text.upper())
+
+
+class _EchoExecutor(Executor):
+    @handler
+    async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
+        await ctx.yield_output(text)
+
+
+def build_upper_workflow() -> Workflow:
+    return WorkflowBuilder(start_executor=_UpperExecutor(id="upper")).build()
+
+
+def build_echo_workflow() -> Workflow:
+    return WorkflowBuilder(start_executor=_EchoExecutor(id="echo")).build()
+
+
+class _MultiChunkExecutor(Executor):
+    """Yields three separate ``output`` events so streaming has something to chew on."""
+
+    @handler
+    async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
+        for chunk in (f"{text}-1", f"{text}-2", f"{text}-3"):
+            await ctx.yield_output(chunk)
+
+
+def build_multi_chunk_workflow() -> Workflow:
+    return WorkflowBuilder(start_executor=_MultiChunkExecutor(id="multi")).build()

python/packages/hosting/tests/hosting/conftest.py

@@ -0,0 +1,25 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Pytest configuration for hosting tests."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+
+def pytest_configure() -> None:
+    """Make workflow fixtures importable in package-local and aggregate test modes."""
+    module_name = "hosting_workflow_fixtures"
+    if module_name in sys.modules:
+        return
+
+    fixture_path = Path(__file__).with_name("_workflow_fixtures.py")
+    spec = importlib.util.spec_from_file_location(module_name, fixture_path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"Unable to load workflow fixtures from {fixture_path}")
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)

python/packages/hosting/tests/hosting/test_host_disk.py

@@ -0,0 +1,239 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for narrowed ``state_dir`` support in :class:`AgentFrameworkHost`."""
+
+from __future__ import annotations
+
+import importlib
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+from agent_framework import AgentSession
+
+from agent_framework_hosting import AgentFrameworkHost, ChannelContext, ChannelContribution
+
+pytest.importorskip("diskcache")
+
+
+class _AgentStub:
+    """Bare-minimum SupportsAgentRun stub for host construction."""
+
+    id = "agent-stub"
+    name: str | None = "Agent Stub"
+    description: str | None = "Test agent stub"
+
+    def create_session(self, *, session_id: str | None = None) -> AgentSession:
+        return AgentSession(session_id=session_id)
+
+    def get_session(self, service_session_id: str, *, session_id: str | None = None) -> AgentSession:
+        return AgentSession(service_session_id=service_session_id, session_id=session_id)
+
+    def run(self, *_args: Any, **_kwargs: Any) -> Any:  # pragma: no cover - unused
+        raise RuntimeError("not invoked")
+
+
+class _ChannelStub:
+    name = "stub"
+    path = "/stub"
+
+    def contribute(self, context: ChannelContext) -> ChannelContribution:
+        del context
+        return ChannelContribution()
+
+
+def _close_host_disk(host: AgentFrameworkHost) -> None:
+    """Release any session-alias store held by ``host``."""
+    if host._sessions_store is not None:
+        host._sessions_store.close()
+
+
+def test_state_dir_none_keeps_plain_alias_dict(tmp_path: Path) -> None:
+    """No store, no alias persistence, no files written."""
+    host = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
+    assert host._sessions_store is None
+    assert isinstance(host._session_aliases, dict)
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_string_state_dir_creates_sessions_subfolder_only(tmp_path: Path) -> None:
+    """Passing a single path expands to ``sessions/`` plus lazy checkpoint path."""
+    host = AgentFrameworkHost(
+        target=_AgentStub(),
+        channels=[_ChannelStub()],
+        state_dir=tmp_path,
+    )
+    try:
+        assert host._sessions_store is not None
+        assert (tmp_path / "sessions").is_dir()
+        assert not (tmp_path / "runner").exists()
+        assert not (tmp_path / "links").exists()
+        # Checkpoint path is derived but not created for agent targets.
+        assert not (tmp_path / "checkpoints").exists()
+    finally:
+        _close_host_disk(host)
+
+
+def test_per_component_session_path(tmp_path: Path) -> None:
+    """Dict form lets callers route session aliases to a specific root."""
+    sessions_dir = tmp_path / "state"
+    host = AgentFrameworkHost(
+        target=_AgentStub(),
+        channels=[_ChannelStub()],
+        state_dir={"sessions": sessions_dir},
+    )
+    try:
+        assert sessions_dir.is_dir()
+        assert host._sessions_store is not None
+        assert host._checkpoint_location is None
+    finally:
+        _close_host_disk(host)
+
+
+@pytest.mark.parametrize("key", ["runner", "links", "active", "identities"])
+def test_removed_state_dir_component_keys_raise(tmp_path: Path, key: str) -> None:
+    """Obsolete follow-up components should fail loudly instead of becoming no-ops."""
+    with pytest.raises(ValueError, match="unknown"):
+        AgentFrameworkHost(
+            target=_AgentStub(),
+            channels=[_ChannelStub()],
+            state_dir=cast(Any, {key: tmp_path / key}),
+        )
+
+
+def test_session_aliases_survive_restart(tmp_path: Path) -> None:
+    """Aliases written on host #1 must be visible to host #2."""
+    state_dir = tmp_path / "state"
+
+    host1 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
+    host1._session_aliases["user-1"] = "sess-abc"
+    host1._session_aliases["user-2"] = "sess-def"
+    _close_host_disk(host1)
+
+    host2 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
+    try:
+        assert host2._session_aliases["user-1"] == "sess-abc"
+        assert host2._session_aliases["user-2"] == "sess-def"
+    finally:
+        _close_host_disk(host2)
+
+
+def _build_simple_workflow() -> Any:
+    """Build a no-op workflow for checkpoint-wiring tests."""
+    build_upper_workflow = importlib.import_module("hosting_workflow_fixtures").build_upper_workflow
+
+    return build_upper_workflow()
+
+
+def test_single_path_state_dir_wires_workflow_checkpoints(tmp_path: Path) -> None:
+    """``state_dir="/foo"`` + workflow target → ``/foo/checkpoints/`` is used."""
+    workflow = _build_simple_workflow()
+    host = AgentFrameworkHost(
+        target=workflow,
+        channels=[_ChannelStub()],
+        state_dir=tmp_path,
+    )
+    try:
+        assert host._checkpoint_location == tmp_path / "checkpoints"
+    finally:
+        _close_host_disk(host)
+
+
+def test_mapping_state_dir_checkpoints_key_wires_workflow_checkpoints(tmp_path: Path) -> None:
+    """``state_dir={"checkpoints": ...}`` + workflow target → that path is used."""
+    workflow = _build_simple_workflow()
+    ckpt_dir = tmp_path / "ck"
+    host = AgentFrameworkHost(
+        target=workflow,
+        channels=[_ChannelStub()],
+        state_dir={"checkpoints": ckpt_dir},
+    )
+    try:
+        assert host._checkpoint_location == ckpt_dir
+        assert host._sessions_store is None
+    finally:
+        _close_host_disk(host)
+
+
+def test_mapping_state_dir_omits_checkpoints_for_workflow(tmp_path: Path) -> None:
+    """Mapping form lets workflow callers opt out of checkpoint persistence."""
+    workflow = _build_simple_workflow()
+    host = AgentFrameworkHost(
+        target=workflow,
+        channels=[_ChannelStub()],
+        state_dir={"sessions": tmp_path / "s"},
+    )
+    try:
+        assert host._checkpoint_location is None
+    finally:
+        _close_host_disk(host)
+
+
+def test_explicit_checkpoint_location_wins_over_state_dir(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    """``checkpoint_location`` + ``state_dir`` → explicit param wins + warn."""
+    workflow = _build_simple_workflow()
+    explicit = tmp_path / "explicit-ck"
+    with caplog.at_level("WARNING", logger="agent_framework.hosting"):
+        host = AgentFrameworkHost(
+            target=workflow,
+            channels=[_ChannelStub()],
+            checkpoint_location=explicit,
+            state_dir=tmp_path,
+        )
+    try:
+        assert host._checkpoint_location == explicit
+        assert any(
+            "state_dir['checkpoints']" in rec.message and "checkpoint_location" in rec.message for rec in caplog.records
+        )
+    finally:
+        _close_host_disk(host)
+
+
+def test_state_dir_checkpoints_for_agent_target_silent_for_single_path(tmp_path: Path) -> None:
+    """Single-path state_dir + agent target → no checkpoint, no warning."""
+    host = AgentFrameworkHost(
+        target=_AgentStub(),
+        channels=[_ChannelStub()],
+        state_dir=tmp_path,
+    )
+    try:
+        assert host._checkpoint_location is None
+        assert not (tmp_path / "checkpoints").exists()
+    finally:
+        _close_host_disk(host)
+
+
+def test_state_dir_checkpoints_for_agent_target_warns_when_explicit(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Mapping form with ``checkpoints`` + agent target → warn."""
+    with caplog.at_level("WARNING", logger="agent_framework.hosting"):
+        host = AgentFrameworkHost(
+            target=_AgentStub(),
+            channels=[_ChannelStub()],
+            state_dir={"checkpoints": tmp_path / "ck"},
+        )
+    try:
+        assert host._checkpoint_location is None
+        assert any(
+            "state_dir['checkpoints']" in rec.message and "not a Workflow" in rec.message for rec in caplog.records
+        )
+    finally:
+        _close_host_disk(host)
+
+
+def test_state_dir_checkpoints_conflicts_with_workflow_own_storage(tmp_path: Path) -> None:
+    """Derived checkpoint path triggers the same conflict guard as explicit."""
+    from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder
+
+    _UpperExecutor = importlib.import_module("hosting_workflow_fixtures")._UpperExecutor
+    workflow = WorkflowBuilder(
+        start_executor=_UpperExecutor(id="upper"),
+        checkpoint_storage=InMemoryCheckpointStorage(),
+    ).build()
+    with pytest.raises(RuntimeError, match="already has checkpoint storage"):
+        AgentFrameworkHost(
+            target=workflow,
+            channels=[_ChannelStub()],
+            state_dir=tmp_path,
+        )

python/packages/hosting/tests/hosting/test_isolation.py

@@ -0,0 +1,316 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the per-request isolation contextvar surface in
+:mod:`agent_framework_hosting._isolation`.
+
+The isolation keys are the ONLY seam Foundry-aware providers use to
+find partition keys, and the host's ASGI middleware lifts them off the
+two well-known headers on every inbound HTTP request. A regression
+that drops the lookup, mistypes a header name, or fails to reset the
+contextvar would silently misroute writes / leak per-request state
+across requests, with zero unit-test signal — so cover the surface
+fully here.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+from agent_framework import AgentSession
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import BaseRoute, Route
+from starlette.testclient import TestClient
+
+from agent_framework_hosting import (
+    AgentFrameworkHost,
+    Channel,
+    ChannelContext,
+    ChannelContribution,
+    IsolationKeys,
+    get_current_isolation_keys,
+    reset_current_isolation_keys,
+    set_current_isolation_keys,
+)
+from agent_framework_hosting._isolation import (  # pyright: ignore[reportPrivateUsage]
+    ISOLATION_HEADER_CHAT,
+    ISOLATION_HEADER_USER,
+    current_isolation_keys,
+)
+
+
+class TestIsolationKeys:
+    def test_defaults_to_none_pair(self) -> None:
+        keys = IsolationKeys()
+        assert keys.user_key is None
+        assert keys.chat_key is None
+        assert keys.is_empty is True
+
+    def test_partial_with_only_user_is_not_empty(self) -> None:
+        keys = IsolationKeys(user_key="alice")
+        assert keys.user_key == "alice"
+        assert keys.chat_key is None
+        assert keys.is_empty is False
+
+    def test_partial_with_only_chat_is_not_empty(self) -> None:
+        keys = IsolationKeys(chat_key="general")
+        assert keys.is_empty is False
+
+    def test_full_pair_is_not_empty(self) -> None:
+        keys = IsolationKeys(user_key="alice", chat_key="general")
+        assert keys.is_empty is False
+
+
+class TestContextVarHelpers:
+    def test_default_is_none(self) -> None:
+        # Each test gets a fresh contextvar value because pytest runs
+        # tests in fresh contexts. ``get`` returns the default.
+        assert get_current_isolation_keys() is None
+
+    def test_set_and_get_round_trip(self) -> None:
+        token = set_current_isolation_keys(IsolationKeys(user_key="alice", chat_key="general"))
+        try:
+            current = get_current_isolation_keys()
+            assert current is not None
+            assert current.user_key == "alice"
+            assert current.chat_key == "general"
+        finally:
+            reset_current_isolation_keys(token)
+        # Reset restores prior value (None in the default context).
+        assert get_current_isolation_keys() is None
+
+    def test_set_with_none_clears(self) -> None:
+        outer = set_current_isolation_keys(IsolationKeys(user_key="alice"))
+        try:
+            inner = set_current_isolation_keys(None)
+            try:
+                assert get_current_isolation_keys() is None
+            finally:
+                reset_current_isolation_keys(inner)
+            # Reset surfaces the outer value again.
+            current = get_current_isolation_keys()
+            assert current is not None
+            assert current.user_key == "alice"
+        finally:
+            reset_current_isolation_keys(outer)
+
+    def test_module_level_contextvar_is_the_same_instance(self) -> None:
+        """Direct contextvar access (used by the ASGI middleware) and the
+        public `get_current_isolation_keys()` helper read from the SAME
+        underlying contextvar. A regression that introduced a second
+        contextvar would silently break the middleware → provider hop."""
+        token = current_isolation_keys.set(IsolationKeys(user_key="bob"))
+        try:
+            via_helper = get_current_isolation_keys()
+            assert via_helper is not None
+            assert via_helper.user_key == "bob"
+        finally:
+            current_isolation_keys.reset(token)
+
+
+class TestHeaderConstants:
+    """The two header names are part of the public contract — they
+    match the ones the Foundry Hosted Agents runtime stamps on every
+    inbound request. A typo here would silently misroute partition
+    writes."""
+
+    def test_user_header_value(self) -> None:
+        assert ISOLATION_HEADER_USER == "x-agent-user-isolation-key"
+
+    def test_chat_header_value(self) -> None:
+        assert ISOLATION_HEADER_CHAT == "x-agent-chat-isolation-key"
+
+
+# --------------------------------------------------------------------------- #
+# End-to-end: ASGI middleware lifts the headers into the contextvar.
+# --------------------------------------------------------------------------- #
+
+
+class _IsolationProbeChannel:
+    """A minimal Channel that exposes a single GET route which captures
+    the contextvar value INSIDE the request and returns it as JSON.
+
+    Tests use this to exercise the full middleware → contextvar →
+    handler hop end-to-end.
+    """
+
+    name = "probe"
+    path = ""
+
+    def __init__(self) -> None:
+        self.captured: list[IsolationKeys | None] = []
+
+        async def _handler(_request: Request) -> JSONResponse:
+            keys = get_current_isolation_keys()
+            self.captured.append(keys)
+            payload: dict[str, str | bool | None]
+            payload = (
+                {"user": keys.user_key, "chat": keys.chat_key}
+                if keys is not None
+                else {"user": None, "chat": None, "_present": False}
+            )
+            return JSONResponse(payload)
+
+        self._routes: list[BaseRoute] = [Route("/probe", _handler)]
+
+    def contribute(self, context: ChannelContext) -> ChannelContribution:
+        del context
+        return ChannelContribution(routes=self._routes)
+
+
+def _make_host_with_probe() -> tuple[AgentFrameworkHost, _IsolationProbeChannel]:
+    class _NoopAgent:
+        id = "noop-agent"
+        name: str | None = "Noop Agent"
+        description: str | None = "Test noop agent"
+
+        def create_session(self, *, session_id: str | None = None) -> AgentSession:
+            return AgentSession(session_id=session_id)
+
+        def get_session(self, service_session_id: str, *, session_id: str | None = None) -> AgentSession:
+            return AgentSession(service_session_id=service_session_id, session_id=session_id)
+
+        def run(self, *_args: object, **_kwargs: object) -> Any:  # pragma: no cover - never called
+            raise RuntimeError("not invoked")
+
+    probe = _IsolationProbeChannel()
+    assert isinstance(probe, Channel)
+    host = AgentFrameworkHost(target=_NoopAgent(), channels=[probe])
+    return host, probe
+
+
+class TestIsolationMiddlewareEndToEnd:
+    def test_headers_ignored_outside_foundry_environment(self) -> None:
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get(
+                "/probe",
+                headers={
+                    ISOLATION_HEADER_USER: "alice-uid",
+                    ISOLATION_HEADER_CHAT: "general-cid",
+                },
+            )
+        assert r.status_code == 200
+        assert r.json() == {"user": None, "chat": None, "_present": False}
+        assert probe.captured == [None]
+
+    def test_both_headers_lifted_into_contextvar(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get(
+                "/probe",
+                headers={
+                    ISOLATION_HEADER_USER: "alice-uid",
+                    ISOLATION_HEADER_CHAT: "general-cid",
+                },
+            )
+        assert r.status_code == 200
+        assert r.json() == {"user": "alice-uid", "chat": "general-cid"}
+        assert len(probe.captured) == 1
+        captured = probe.captured[0]
+        assert captured is not None
+        assert captured.user_key == "alice-uid"
+        assert captured.chat_key == "general-cid"
+
+    def test_only_user_header_lifted(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """One-header-only branch: the middleware still binds (chat=None)."""
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
+        assert r.status_code == 200
+        assert r.json() == {"user": "alice-uid", "chat": None}
+
+    def test_only_chat_header_lifted(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get("/probe", headers={ISOLATION_HEADER_CHAT: "general-cid"})
+        assert r.status_code == 200
+        assert r.json() == {"user": None, "chat": "general-cid"}
+
+    def test_no_headers_keeps_contextvar_none(self) -> None:
+        """Local-dev path: with neither header present the middleware is
+        a no-op and the contextvar stays at its default ``None`` —
+        providers see "no isolation" and route to the in-memory
+        fallback rather than picking up stale per-request state."""
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get("/probe")
+        assert r.status_code == 200
+        assert r.json() == {"user": None, "chat": None, "_present": False}
+        assert probe.captured == [None]
+
+    def test_empty_header_value_treated_as_absent(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """A header that's present but empty must not bind an empty key —
+        ``IsolationContext`` rejects empty strings on the read side."""
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r = client.get(
+                "/probe",
+                headers={
+                    ISOLATION_HEADER_USER: "",
+                    ISOLATION_HEADER_CHAT: "general-cid",
+                },
+            )
+        assert r.status_code == 200
+        # Empty user header decodes to None; chat key stays bound.
+        assert r.json() == {"user": None, "chat": "general-cid"}
+
+    def test_contextvar_resets_after_request(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """The middleware must call ``reset_current_isolation_keys`` in
+        a ``finally`` so per-request state never leaks across requests
+        or back into the calling thread's context."""
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+        with TestClient(host.app) as client:  # type: ignore[attr-defined]
+            r1 = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
+            assert r1.status_code == 200
+            # Reading the contextvar OUTSIDE the request scope must see
+            # the default — not the value the prior request bound.
+            assert get_current_isolation_keys() is None
+            # And a follow-up request without headers gets a clean
+            # ``None`` rather than inheriting alice-uid.
+            r2 = client.get("/probe")
+            assert r2.json() == {"user": None, "chat": None, "_present": False}
+
+    def test_concurrent_requests_get_isolated_contextvars(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Different requests run in different async contexts; binding
+        from request A must NOT leak into a concurrent request B."""
+        monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
+        host, probe = _make_host_with_probe()
+
+        async def _drive() -> None:
+            # Run two requests in parallel asyncio tasks against the
+            # same TestClient and assert their captures don't bleed
+            # into each other.
+            async def _hit(user_key: str) -> dict[str, str | None]:
+                with TestClient(host.app) as client:  # type: ignore[attr-defined]
+                    r = client.get("/probe", headers={ISOLATION_HEADER_USER: user_key})
+                return r.json()  # type: ignore[no-any-return]
+
+            r_alice, r_bob = await asyncio.gather(_hit("alice-uid"), _hit("bob-uid"))
+            assert r_alice == {"user": "alice-uid", "chat": None}
+            assert r_bob == {"user": "bob-uid", "chat": None}
+
+        asyncio.run(_drive())
+
+
+class TestNonHttpScopesPassThrough:
+    """The middleware intentionally only inspects ``http`` scopes;
+    lifespan / websocket scopes are forwarded untouched. A regression
+    that touched lifespan scopes here would crash boot."""
+
+    async def test_lifespan_scope_does_not_consult_headers(self) -> None:
+        # The TestClient context manager exercises the lifespan scope
+        # implicitly; if the middleware tried to decode headers on a
+        # non-http scope this would raise. Exercise it without binding
+        # any contextvar work.
+        host, _probe = _make_host_with_probe()
+        with TestClient(host.app):  # type: ignore[attr-defined]
+            # Just enter / exit; no requests.
+            pass

python/packages/hosting/tests/hosting/test_types.py

@@ -0,0 +1,50 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the channel-neutral envelope types in :mod:`agent_framework_hosting._types`."""
+
+from __future__ import annotations
+
+from agent_framework_hosting import (
+    ChannelIdentity,
+    ChannelRequest,
+    ChannelSession,
+)
+
+
+class TestChannelRequest:
+    def test_required_fields_only(self) -> None:
+        req = ChannelRequest(channel="responses", operation="message.create", input="hi")
+        assert req.channel == "responses"
+        assert req.operation == "message.create"
+        assert req.input == "hi"
+        assert req.session is None
+        assert req.options is None
+        assert req.session_mode == "auto"
+        assert req.metadata == {}
+        assert req.attributes == {}
+        assert req.stream is False
+        assert req.identity is None
+
+    def test_with_session_and_identity(self) -> None:
+        req = ChannelRequest(
+            channel="telegram",
+            operation="message.create",
+            input="hi",
+            session=ChannelSession(isolation_key="user:42"),
+            identity=ChannelIdentity(channel="telegram", native_id="42"),
+        )
+        assert req.session is not None
+        assert req.session.isolation_key == "user:42"
+        assert req.identity is not None
+        assert req.identity.channel == "telegram"
+        assert req.identity.native_id == "42"
+
+
+class TestChannelIdentity:
+    def test_attributes_default_empty_mapping(self) -> None:
+        ident = ChannelIdentity(channel="teams", native_id="abc")
+        assert dict(ident.attributes) == {}
+
+    def test_attributes_passthrough(self) -> None:
+        ident = ChannelIdentity(channel="teams", native_id="abc", attributes={"role": "user"})
+        assert dict(ident.attributes) == {"role": "user"}

python/packages/tools/tests/test_docker_shell_tool.py

@@ -10,6 +10,7 @@ available in CI / dev sandboxes).
 
 from __future__ import annotations
 
+import subprocess
 import sys
 
 import pytest
@@ -24,11 +25,27 @@ from agent_framework_tools.shell._docker import (
     build_run_argv,
 )
 
+
+def _docker_image_available(image: str) -> bool:
+    if not is_docker_available():
+        return False
+    try:
+        result = subprocess.run(
+            ["docker", "image", "inspect", image],
+            capture_output=True,
+            check=False,
+            timeout=5.0,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return False
+    return result.returncode == 0
+
+
 # Integration tests use Linux container images (alpine) that don't run
 # under Docker Desktop's default Windows-container mode.
 _skip_if_no_linux_docker = pytest.mark.skipif(
-    not is_docker_available() or sys.platform == "win32",
-    reason="docker daemon unavailable or running Windows containers",
+    not _docker_image_available("alpine:3") or sys.platform == "win32",
+    reason="docker daemon unavailable, alpine:3 image missing, or running Windows containers",
 )
 
 # --------------------------------------------------------------------- argv builders

python/pyproject.toml

@@ -91,6 +91,8 @@ agent-framework-foundry-hosting = { workspace = true }
 agent-framework-foundry-local = { workspace = true }
 agent-framework-gemini = { workspace = true }
 agent-framework-github-copilot = { workspace = true }
+agent-framework-hosting = { workspace = true }
+agent-framework-hosting-responses = { workspace = true }
 agent-framework-hyperlight = { workspace = true }
 agent-framework-lab = { workspace = true }
 agent-framework-mem0 = { workspace = true }

python/samples/01-get-started/05_functional_workflow_with_agents.py

@@ -12,6 +12,10 @@ import asyncio
 from agent_framework import Agent, workflow
 from agent_framework.foundry import FoundryChatClient
 from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
+
+# Load environment variables from .env file (e.g., FOUNDRY_PROJECT_ENDPOINT, FOUNDRY_MODEL)
+load_dotenv()
 
 # <create_agents>
 client = FoundryChatClient(credential=AzureCliCredential())

python/samples/02-agents/chat_client/chat_response_cancellation.py

@@ -24,8 +24,9 @@ async def main() -> None:
     Creates a task for the chat request, waits briefly, then cancels it to show proper cleanup.
 
     Configuration:
-    - OpenAI model ID: Use "model" parameter or "OPENAI_MODEL" environment variable
-    - OpenAI API key: Use "api_key" parameter or "OPENAI_API_KEY" environment variable
+    - FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint URL
+    - FOUNDRY_MODEL: Model deployment name (e.g. gpt-4o)
+    - Authentication: Run `az login` to authenticate via AzureCliCredential
     """
     client = FoundryChatClient(credential=AzureCliCredential())
 

python/samples/04-hosting/af-hosting/README.md

@@ -0,0 +1,42 @@
+# Multi-channel hosting samples
+
+End-to-end samples for serving an `agent-framework` agent (or workflow)
+through one or more **channels** with `agent-framework-hosting`.
+
+The general hosting plumbing lives in
+[`agent-framework-hosting`](../../../packages/hosting); each channel is
+its own package. This first sample set includes
+`agent-framework-hosting-responses`.
+
+| Sample | What it shows | Packaging |
+|---|---|---|
+| [`local_responses/`](./local_responses) | The minimal shape: one agent + one `@tool` + `ResponsesChannel` + a single `run_hook` that strips caller-supplied options and forces a `reasoning` preset. | **Local only.** Start here to learn the run-hook seam. |
+| [`local_responses_workflow/`](./local_responses_workflow) | A 4-step `Workflow` (typed `SloganBrief` intake → writer → legal → formatter) hosted behind the Responses channel via a `run_hook` that parses inbound text/JSON into the workflow's typed input. The host writes per-conversation checkpoints via `checkpoint_location=…`. Demonstrates workflow targets + structured input adaptation + resume-across-turns. Includes a `call_server.rest` file with REST examples. | **Local only.** |
+
+Each sample is fully self-contained — its own `pyproject.toml`, `uv.lock`,
+server `app.py`, calling script(s), and `storage/` directory. Every
+sample uses `[tool.uv.sources]` to wire its `agent-framework-hosting*`
+dependencies to the
+[`main`](https://github.com/microsoft/agent-framework/tree/main)
+branch of the upstream repo via git refs, so they install cleanly outside
+the monorepo while the hosting packages are still pre-PyPI. Once those
+packages publish, drop the `[tool.uv.sources]` block and let the
+declared deps resolve from PyPI.
+
+## Relationship to `../foundry-hosted-agents/`
+
+The sibling [`../foundry-hosted-agents/`](../foundry-hosted-agents) directory
+contains samples for the **`agent-framework-hosted`** stack — agents
+that run **inside** the Foundry Hosted Agents platform using its
+built-in protocol surface (Responses, Invocations, conversation store,
+isolation, identity), with **no `agent-framework-hosting` package
+involved**.
+
+| Aspect | `af-hosting/` (this directory) | `foundry-hosted-agents/` |
+|---|---|---|
+| Server stack | `agent-framework-hosting` + `agent-framework-hosting-responses` | `agent-framework-hosted` only — the Foundry Hosted Agents runtime owns the HTTP surface |
+| Channels | Responses only in this initial sample set | The platform exposes Responses + Invocations |
+| Run target | Local Hypercorn (`local_responses/`, `local_responses_workflow/`) | Hosted Agents *or* local container; targets the Hosted Agents platform contract |
+| When to pick this | You want to learn the host/channel seams locally or need custom hosting middleware | You want zero hosting boilerplate, leveraging the Foundry-managed surface |
+
+The table above summarizes the cross-sample story.

python/samples/04-hosting/af-hosting/local_responses/README.md

@@ -0,0 +1,54 @@
+# local_responses — Responses-only with a settings-altering hook
+
+The smallest end-to-end `agent-framework-hosting` shape: one Foundry
+agent with a `@tool`, one `ResponsesChannel`, one `run_hook`. Useful as
+the entry-point sample for understanding the **channel run-hook** seam
+without any multi-channel or identity-link concerns.
+
+What the run hook demonstrates:
+
+- **Strips** caller-supplied `model` / `temperature` / `store` so the
+  host owns the backing deployment and persistence settings.
+- **Forces** a `reasoning` preset (`effort=medium`, `summary=auto`) on
+  every turn — caller-side overrides are ignored.
+
+`app:app` is a module-level Starlette ASGI app; recommended local launch
+is Hypercorn.
+
+## Run
+
+```bash
+export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
+export FOUNDRY_MODEL=gpt-5-nano
+az login
+
+uv sync
+uv run hypercorn app:app --bind 0.0.0.0:8000
+```
+
+Single-process for quick iteration:
+
+```bash
+uv run python app.py
+```
+
+## Call locally
+
+```bash
+uv sync --group dev
+
+# Plain OpenAI SDK call:
+uv run python call_server.py
+
+# The client intentionally omits `model`; the host chooses the backing
+# deployment from FOUNDRY_MODEL.
+
+# The script then sends a second turn, "And what about Amsterdam?",
+# using the first `response.id` as `previous_response_id`.
+
+# Same two-turn interaction through an Agent Framework Agent backed by
+# OpenAIChatClient, with streaming enabled:
+uv run python call_server_af.py
+```
+
+> This sample is **local-only** — no Dockerfile, no Foundry packaging.

python/samples/04-hosting/af-hosting/local_responses/app.py

@@ -0,0 +1,126 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Minimal Responses-only hosting sample.
+
+Single agent with one ``@tool`` (``lookup_weather``), single channel
+(``ResponsesChannel``), one ``run_hook`` that demonstrates the
+settings-mutation seam over caller-supplied options.
+
+What the hook does
+------------------
+On every Responses request the hook receives the ``ChannelRequest`` that
+the channel built from the inbound HTTP body. It:
+
+- strips ``model`` (the host owns the backing deployment), ``store``
+  (this agent owns persistence), and ``temperature`` (the configured
+  model may not honor it),
+- forces a ``reasoning`` effort + summary preset so the deployed surface
+  is consistent regardless of what the caller sent.
+
+The hook is the documented escape hatch over the uniform
+``ChannelRequest`` envelope.
+
+Run
+---
+``app`` is a module-level Starlette ASGI app. Recommended local launch::
+
+    uv sync
+    az login
+    export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
+    export FOUNDRY_MODEL=gpt-5-nano
+    uv run hypercorn app:app --bind 0.0.0.0:8000
+
+Or use the ``__main__`` block (single-process Hypercorn) for quick
+iteration::
+
+    uv run python app.py
+
+Then call it::
+
+    uv run python call_server.py "What is the weather in Tokyo?"
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import replace
+from pathlib import Path
+from typing import Annotated
+
+from agent_framework import Agent, FileHistoryProvider, tool
+from agent_framework_foundry import FoundryChatClient
+from agent_framework_hosting import AgentFrameworkHost, ChannelRequest
+from agent_framework_hosting_responses import ResponsesChannel
+from azure.identity.aio import DefaultAzureCredential
+
+SESSIONS_DIR = Path(__file__).resolve().parent / "storage" / "sessions"
+SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+@tool(approval_mode="never_require")
+def lookup_weather(
+    location: Annotated[str, "The city to look up weather for."],
+) -> str:
+    """Return a deterministic weather report for a city."""
+    high_temp = 5 + (sum(location.encode("utf-8")) % 21)
+    reports = {
+        "Seattle": f"Seattle is rainy with a high of {high_temp}°C.",
+        "Amsterdam": f"Amsterdam is cloudy with a high of {high_temp}°C.",
+        "Tokyo": f"Tokyo is clear with a high of {high_temp}°C.",
+    }
+    return reports.get(location, f"{location} is sunny with a high of {high_temp}°C.")
+
+
+# the run hook defines what you want to allow the user to passthrough when they call your host
+# since the responses clients can call with all of the responses options,
+# you can decide with this run_hook which of those: are rejected
+# which are passed through, which are altered, which are added.
+# In this sample below, we are removing, model, temperature and store if set
+# and we add reasoning, but note that this could also be set on the Agent itself
+# the difference is that this option is specific to the Responses channel
+# so if you want to differentiate between options over channels
+# you would set the option in the run_hook, if it needs to be the same (like store)
+# you would set it in the agent.
+def run_hook(request: ChannelRequest, **_: object) -> ChannelRequest:
+    """Strip caller-supplied options the host should own and force a
+    reasoning preset."""
+    options = dict(request.options or {})
+
+    # The host owns the backing deployment; the agent's default_options
+    # own ``store``; the model may not honor ``temperature``. Strip them
+    # so the caller can't override.
+    options.pop("model", None)
+    options.pop("temperature", None)
+    options.pop("store", None)
+
+    # Force a consistent reasoning preset on every turn.
+    options["reasoning"] = {"effort": "medium", "summary": "auto"}
+
+    return replace(request, options=options or None)
+
+
+def build_host() -> AgentFrameworkHost:
+    # Here we define how our agent should run, with tools, options, etc:
+    agent = Agent(
+        client=FoundryChatClient(credential=DefaultAzureCredential()),
+        name="WeatherAgent",
+        instructions=(
+            "You are a friendly weather assistant. Use the lookup_weather tool "
+            "for any weather question and answer in one short sentence."
+        ),
+        tools=[lookup_weather],
+        context_providers=[FileHistoryProvider(SESSIONS_DIR)],
+        default_options={"store": False},
+    )
+    return AgentFrameworkHost(
+        target=agent,
+        channels=[ResponsesChannel(run_hook=run_hook)],
+        debug=True,
+    )
+
+
+app = build_host().app
+
+
+if __name__ == "__main__":
+    build_host().serve(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))

python/samples/04-hosting/af-hosting/local_responses/call_server.py

@@ -0,0 +1,51 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Local client for the local_responses sample.
+
+Posts to ``/responses`` using the standard ``openai`` SDK.
+
+Pass ``--previous-response-id <id>`` to continue a conversation by its
+``response.id`` (returned in the prior response).
+
+Start the server first (in another shell)::
+
+    uv run python app.py
+
+Then::
+
+    uv run python call_server.py
+
+The script sends a follow-up turn ("And what about Amsterdam?") using the
+first response's ``response.id`` as ``previous_response_id``.
+"""
+
+from __future__ import annotations
+
+from openai import OpenAI
+
+BASE_URL = "http://127.0.0.1:8000"
+PROMPT = "What is the weather in Tokyo?"
+FOLLOW_UP_PROMPT = "And what about Amsterdam?"
+
+
+def main() -> None:
+    client = OpenAI(base_url=BASE_URL, api_key="not-needed")
+    response = client.responses.create(
+        input=PROMPT,
+    )
+    print(f"User: {PROMPT}")
+    print(f"Agent: {response.output_text}")
+    print(f"Response ID: {response.id}")
+
+    follow_up = client.responses.create(
+        input=FOLLOW_UP_PROMPT,
+        previous_response_id=response.id,
+    )
+    print()
+    print(f"User: {FOLLOW_UP_PROMPT}")
+    print(f"Agent: {follow_up.output_text}")
+    print(f"Response ID: {follow_up.id}")
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/af-hosting/local_responses/call_server_af.py

@@ -0,0 +1,60 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Agent Framework agent client for the local_responses sample.
+
+Creates a local :class:`agent_framework.Agent` backed by
+:class:`agent_framework.openai.OpenAIChatClient`, points that client at the
+hosted ``/responses`` endpoint, and streams both turns:
+
+1. ``What is the weather in Tokyo?``
+2. ``And what about Amsterdam?``
+
+Both turns use the same :class:`agent_framework.AgentSession`; the first
+turn binds the hosted response id to the session, and the second turn
+continues through that session.
+
+Start the server first (in another shell)::
+
+    uv run python app.py
+
+Then::
+
+    uv run python call_server_af.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from agent_framework import Agent
+from agent_framework.openai import OpenAIChatClient
+
+BASE_URL = "http://127.0.0.1:8000"
+PROMPTS = [
+    "What is the weather in Tokyo?",
+    "And what about Amsterdam?",
+]
+
+
+async def main() -> None:
+    agent = Agent(
+        client=OpenAIChatClient(base_url=BASE_URL, api_key="not-needed"),
+        name="HostedWeatherClient",
+    )
+    session = agent.create_session()
+
+    for prompt in PROMPTS:
+        print(f"User: {prompt}")
+        stream = agent.run(prompt, stream=True, session=session)
+        print("Agent: ", end="", flush=True)
+        async for update in stream:
+            if update.text:
+                print(update.text, end="", flush=True)
+
+        response = await stream.get_final_response()
+        print("\n")
+        print(f"Response ID: {response.response_id}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/04-hosting/af-hosting/local_responses/pyproject.toml

@@ -0,0 +1,27 @@
+[project]
+name = "agent-framework-hosting-sample-local-responses"
+version = "0.0.1"
+description = "Minimal Responses-only local hosting sample with a settings-altering run hook."
+requires-python = ">=3.10"
+dependencies = [
+    "agent-framework-foundry",
+    "agent-framework-hosting",
+    "agent-framework-hosting-responses",
+    "azure-identity",
+    "aiohttp>=3.13.5",
+    "hypercorn>=0.17",
+]
+
+[dependency-groups]
+dev = [
+    "agent-framework-openai",
+    "openai>=1.99",
+]
+
+[tool.uv]
+package = false
+
+[tool.uv.sources]
+agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" }
+agent-framework-hosting-responses = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-responses" }
+agent-framework-openai = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/openai" }

python/samples/04-hosting/af-hosting/local_responses_workflow/README.md

@@ -0,0 +1,83 @@
+# local_responses_workflow — workflow target with run-hook prep + checkpoints
+
+A `Workflow` (writer → legal reviewer → formatter) hosted
+behind the **Responses API**, with the host configured to
+**persist per-conversation checkpoints**. Mirrors
+[`../../foundry-hosted-agents/responses/05_workflows/`](../../foundry-hosted-agents/responses/05_workflows/)
+but uses the `agent-framework-hosting` stack instead of the
+Foundry-Hosted-Agents runtime. The `run_hook` prepares the writer prompt
+before the workflow starts.
+
+## What's interesting
+
+- `AgentFrameworkHost(target=workflow, …)` — the host detects a
+  `Workflow` target and dispatches to `workflow.run(...)` (no
+  `Agent.create_session(...)`).
+- `ResponsesChannel` is mounted at `/responses` with a `prepare_writer_prompt`
+  run hook that **adapts the channel-native input into the workflow start
+  executor's input**. Responses delivers a `list[Message]`; the hook normalises
+  it to text and prepares the prompt the writer agent receives.
+- The hook parses the inbound text as JSON
+  (`{"topic": ..., "style": ..., "audience": ...}`); if parsing fails
+  it uses the whole text as `topic` with defaults.
+- The workflow starts directly at the writer `AgentExecutor`; no extra intake
+  executor is needed because the hook performs the one preparation step.
+- `checkpoint_location=storage/checkpoints/` — the host scopes a
+  `FileCheckpointStorage` per conversation (Responses keys it on
+  `previous_response_id` / `conversation_id`) and **restores from the
+  latest checkpoint at the start of every turn** before applying the new
+  input. Without an isolation key the host skips checkpointing for that request.
+- No `HistoryProvider` — the workflow owns its own state via the
+  checkpoint store.
+
+## Run
+
+```bash
+export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
+export FOUNDRY_MODEL=gpt-5-nano
+az login
+
+uv sync
+uv run hypercorn app:app --bind 0.0.0.0:8000
+```
+
+Single-process for quick iteration:
+
+```bash
+uv run python app.py
+```
+
+## Call locally
+
+Two clients are provided next to `app.py`:
+
+- **`call_server.py`** — Python client using the OpenAI SDK (Responses
+  API only).
+- **`call_server.rest`** — raw REST examples for the Responses endpoint
+  (open in VS Code with the REST Client extension or any compatible HTTP-file
+  runner).
+
+```bash
+uv sync --group dev
+
+# Structured brief via the OpenAI SDK (Responses API):
+uv run python call_server.py \
+    '{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
+
+# The client intentionally omits `model`; the host chooses the backing
+# deployment from FOUNDRY_MODEL.
+
+# Plain topic (style/audience default to "modern" / "general"):
+uv run python call_server.py "electric SUV"
+
+# Continue an existing conversation by its `response.id`:
+uv run python call_server.py --previous-response-id <response-id> \
+    '{"topic": "electric SUV", "style": "retro", "audience": "boomers"}'
+```
+
+After a few turns, inspect `storage/checkpoints/<isolation_key>/` —
+each conversation has its own subdirectory of checkpoint files written
+by the host.
+
+> This sample is **local-only** — no Dockerfile, no Foundry packaging.
+> A Foundry-Hosted-Agents-compatible packaging sample will be added separately.

python/samples/04-hosting/af-hosting/local_responses_workflow/app.py

@@ -0,0 +1,182 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Hosted workflow sample with run-hook input prep + checkpoint location.
+
+Same three-agent slogan workflow as
+``../../foundry-hosted-agents/responses/05_workflows/main.py`` (writer →
+legal reviewer → formatter), driven through the ``agent-framework-hosting``
+stack instead of the Foundry-Hosted-Agents runtime.
+
+Workflow shape
+--------------
+``writer`` → ``legal_reviewer`` → ``formatter``. A single run hook parses
+the Responses input and prepares the prompt the writer agent receives.
+
+What this sample shows
+----------------------
+- A :class:`~agent_framework.Workflow` is a valid hosting target — the
+  host detects it and dispatches to ``workflow.run(...)`` instead of
+  ``agent.run(...)``.
+- ``ResponsesChannel(run_hook=...)`` is the seam for **adapting the
+  channel-native input into the workflow start executor's input**.
+  The hook here parses the inbound text as JSON
+  (``{"topic": ..., "style": ..., "audience": ...}``) — if parsing
+  fails it falls back to using the whole text as ``topic`` with
+  defaults — and replaces ``ChannelRequest.input`` with the prepared
+  writer prompt.
+- ``AgentFrameworkHost(checkpoint_location=...)`` enables
+  per-conversation workflow checkpointing. The host scopes the
+  checkpoint storage by ``ChannelRequest.session.isolation_key``
+  (Responses uses ``previous_response_id`` / ``conversation_id`` as the
+  isolation key), and restores from the latest checkpoint before each
+  new turn — so a multi-turn workflow can resume across requests.
+- No ``HistoryProvider`` is configured: the workflow owns its own state
+  via the checkpoint store; the agent-history seam is for plain
+  ``SupportsAgentRun`` agents.
+
+Run
+---
+``app`` is a module-level Starlette ASGI app::
+
+    uv sync
+    az login
+    export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
+    export FOUNDRY_MODEL=gpt-5-nano
+    uv run hypercorn app:app --bind 0.0.0.0:8000
+
+Or for quick iteration::
+
+    uv run python app.py
+
+Then call it with a structured brief::
+
+    uv run python call_server.py \\
+        '{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
+
+Or with just a topic — the hook fills in defaults::
+
+    uv run python call_server.py "Create a slogan for an electric SUV."
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import replace
+from pathlib import Path
+
+from agent_framework import (
+    Agent,
+    AgentExecutor,
+    Message,
+    WorkflowBuilder,
+)
+from agent_framework_foundry import FoundryChatClient
+from agent_framework_hosting import AgentFrameworkHost, ChannelRequest
+from agent_framework_hosting_responses import ResponsesChannel
+from azure.identity.aio import DefaultAzureCredential
+
+CHECKPOINTS_DIR = Path(__file__).resolve().parent / "storage" / "checkpoints"
+CHECKPOINTS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def prepare_writer_prompt(request: ChannelRequest, **_: object) -> ChannelRequest:
+    """Prepare the workflow's initial writer prompt from Responses input.
+
+    The channel hands the host either a ``str`` (rare on the Responses
+    surface) or a list of :class:`Message`. This hook collapses that
+    input to text, accepts either JSON or plain text, and replaces the
+    request input with a plain prompt for the writer executor.
+    """
+
+    def extract_text(value: object) -> str:
+        if isinstance(value, str):
+            return value
+        if isinstance(value, Message):
+            return value.text
+        if isinstance(value, list):
+            return "\n".join(extract_text(item) for item in value)
+        return ""
+
+    text = extract_text(request.input).strip()
+    topic = text or "a generic product"
+    style = "modern"
+    audience = "general"
+    if topic.startswith("{"):
+        try:
+            data = json.loads(topic)
+        except json.JSONDecodeError:
+            data = None
+        if isinstance(data, dict) and "topic" in data:
+            topic = str(data["topic"])
+            style = str(data.get("style", style))
+            audience = str(data.get("audience", audience))
+
+    prompt = (
+        f"Topic: {topic}\n"
+        f"Style: {style}\n"
+        f"Audience: {audience}\n\n"
+        "Write a single short slogan that fits the topic, style, and audience."
+    )
+    return replace(request, input=prompt)
+
+
+def build_host() -> AgentFrameworkHost:
+    client = FoundryChatClient(credential=DefaultAzureCredential())
+
+    writer = Agent(
+        client=client,
+        name="writer",
+        instructions=("You are an excellent slogan writer. You create new slogans based on the given topic."),
+    )
+    legal = Agent(
+        client=client,
+        name="legal_reviewer",
+        instructions=(
+            "You are an excellent legal reviewer. "
+            "Make necessary corrections to the slogan so that it is legally compliant."
+        ),
+    )
+    formatter = Agent(
+        client=client,
+        name="formatter",
+        instructions=(
+            "You are an excellent content formatter. "
+            "You take the slogan and format it in a cool retro style when printing to a terminal."
+        ),
+    )
+
+    # ``context_mode="last_agent"`` ensures each agent only sees the
+    # previous executor's output — matching the Foundry sample.
+    writer_ex = AgentExecutor(writer, context_mode="last_agent")
+    legal_ex = AgentExecutor(legal, context_mode="last_agent")
+    format_ex = AgentExecutor(formatter, context_mode="last_agent")
+
+    workflow = (
+        WorkflowBuilder(
+            start_executor=writer_ex,
+            output_executors=[format_ex],
+        )
+        .add_edge(writer_ex, legal_ex)
+        .add_edge(legal_ex, format_ex)
+        .build()
+    )
+
+    return AgentFrameworkHost(
+        target=workflow,
+        channels=[
+            ResponsesChannel(run_hook=prepare_writer_prompt),
+        ],
+        # The host writes a per-conversation FileCheckpointStorage rooted
+        # at ``CHECKPOINTS_DIR / <isolation_key>`` and restores from the
+        # latest checkpoint at the start of every turn.
+        checkpoint_location=CHECKPOINTS_DIR,
+        debug=True,
+    )
+
+
+app = build_host().app
+
+
+if __name__ == "__main__":
+    build_host().serve(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))

python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.py

@@ -0,0 +1,53 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Local client for the local_responses_workflow sample.
+
+The server expects a structured slogan brief. You can either pass a
+JSON object or a plain topic string (the server's run hook fills the
+other fields with defaults).
+
+Pass ``--previous-response-id <id>`` to continue a conversation by its
+``response.id`` — the host uses that as the workflow checkpoint scope
+key, so the workflow resumes from where it left off.
+
+Start the server first (in another shell)::
+
+    uv run python app.py
+
+Then::
+
+    uv run python call_server.py \\
+        '{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
+
+    uv run python call_server.py "electric SUV"  # uses default style/audience
+"""
+
+from __future__ import annotations
+
+import sys
+
+from openai import OpenAI
+
+BASE_URL = "http://127.0.0.1:8000"
+
+
+def main() -> None:
+    args = sys.argv[1:]
+    previous_response_id: str | None = None
+    if len(args) >= 2 and args[0] == "--previous-response-id":
+        previous_response_id = args[1]
+        args = args[2:]
+        print(f"Resuming response: {previous_response_id}")
+    prompt = " ".join(args) or '{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
+    client = OpenAI(base_url=BASE_URL, api_key="not-needed")
+    response = client.responses.create(
+        input=prompt,
+        previous_response_id=previous_response_id,
+    )
+    print(f"User: {prompt}")
+    print(f"Agent: {response.output_text}")
+    print(f"response.id: {response.id}")
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/af-hosting/local_responses_workflow/call_server.rest

@@ -0,0 +1,48 @@
+# local_responses_workflow — REST examples
+#
+# Use with the VS Code "REST Client" extension (humao.rest-client) or
+# JetBrains HTTP Client. Each `###` block is one request.
+#
+# Start the server in another shell first:
+#   uv run python app.py
+
+@host = http://127.0.0.1:8000
+
+###
+# 1. Responses API — structured brief
+POST {{host}}/responses
+Content-Type: application/json
+
+{
+  "model": "agent",
+  "input": "{\"topic\": \"electric SUV\", \"style\": \"playful\", \"audience\": \"young families\"}"
+}
+
+###
+# 2. Responses API — plain topic, defaults applied by the run hook
+POST {{host}}/responses
+Content-Type: application/json
+
+{
+  "model": "agent",
+  "input": "vintage espresso machine"
+}
+
+###
+# 3. Responses API — continue the conversation by previous_response_id
+#    Replace <RESPONSE_ID> with `id` from one of the responses above —
+#    the host uses it as the workflow checkpoint scope key, so the
+#    workflow resumes from its latest checkpoint before applying the
+#    new input.
+POST {{host}}/responses
+Content-Type: application/json
+
+{
+  "model": "agent",
+  "previous_response_id": "<RESPONSE_ID>",
+  "input": "{\"topic\": \"electric SUV\", \"style\": \"retro\", \"audience\": \"boomers\"}"
+}
+
+###
+# 4. Readiness probe
+GET {{host}}/readiness

python/samples/04-hosting/af-hosting/local_responses_workflow/pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "agent-framework-hosting-sample-local-responses-workflow"
+version = "0.0.1"
+description = "Local hosting sample exposing a 3-agent workflow over the Responses API with per-conversation checkpoint storage."
+requires-python = ">=3.10"
+dependencies = [
+    "agent-framework-foundry",
+    "agent-framework-hosting",
+    "agent-framework-hosting-responses",
+    "azure-identity",
+    "aiohttp>=3.13.5",
+    "hypercorn>=0.17",
+]
+
+[dependency-groups]
+dev = ["openai>=1.99"]
+
+[tool.uv]
+package = false
+
+[tool.uv.sources]
+agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" }
+agent-framework-hosting-responses = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-responses" }

python/uv.lock

@@ -46,6 +46,8 @@ members = [
     "agent-framework-foundry-local",
     "agent-framework-gemini",
     "agent-framework-github-copilot",
+    "agent-framework-hosting",
+    "agent-framework-hosting-responses",
     "agent-framework-hyperlight",
     "agent-framework-lab",
     "agent-framework-mem0",
@@ -624,6 +626,57 @@ requires-dist = [
     { name = "github-copilot-sdk", marker = "python_full_version >= '3.11'", specifier = "==1.0.2" },
 ]
 
+[[package]]
+name = "agent-framework-hosting"
+version = "1.0.0a260424"
+source = { editable = "packages/hosting" }
+dependencies = [
+    { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "starlette", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+]
+
+[package.optional-dependencies]
+disk = [
+    { name = "diskcache", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+]
+serve = [
+    { name = "hypercorn", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+]
+
+[package.dev-dependencies]
+dev = [
+    { name = "httpx", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+]
+
+[package.metadata]
+requires-dist = [
+    { name = "agent-framework-core", editable = "packages/core" },
+    { name = "diskcache", marker = "extra == 'disk'", specifier = ">=5.6" },
+    { name = "hypercorn", marker = "extra == 'serve'", specifier = ">=0.17" },
+    { name = "starlette", specifier = ">=0.37" },
+]
+provides-extras = ["serve", "disk"]
+
+[package.metadata.requires-dev]
+dev = [{ name = "httpx", specifier = ">=0.28.1" }]
+
+[[package]]
+name = "agent-framework-hosting-responses"
+version = "1.0.0a260424"
+source = { editable = "packages/hosting-responses" }
+dependencies = [
+    { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "openai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
+]
+
+[package.metadata]
+requires-dist = [
+    { name = "agent-framework-core", editable = "packages/core" },
+    { name = "agent-framework-hosting", editable = "packages/hosting" },
+    { name = "openai", specifier = ">=1.99.0,<3" },
+]
+
 [[package]]
 name = "agent-framework-hyperlight"
 version = "1.0.0b260521"
@@ -2253,6 +2306,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c7/26/4a2bad8eb430d8d805a4642c4bff25103a37548d74ab346f8b1e024abcc5/deepdiff-9.1.0-py3-none-any.whl", hash = "sha256:80c0460e1993b04f6f0ca79abf25548b129fd218478c4ebb08f80560f5d10610", size = 184662 },
 ]
 
+[[package]]
+name = "diskcache"
+version = "5.6.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/3f/21/1c1ffc1a039ddcc459db43cc108658f32c57d271d7289a2794e401d0fdb6/diskcache-5.6.3.tar.gz", hash = "sha256:2c3a3fa2743d8535d832ec61c2054a1641f41775aa7c556758a109941e33e4fc", size = 67916, upload-time = "2023-08-31T06:12:00.316Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3f/27/4570e78fc0bf5ea0ca45eb1de3818a23787af9b390c0b0a0033a1b8236f9/diskcache-5.6.3-py3-none-any.whl", hash = "sha256:5e31b2d5fbad117cc363ebaf6b689474db18a1f6438bc82358b024abd4c2ca19", size = 45550, upload-time = "2023-08-31T06:11:58.822Z" },
+]
+
 [[package]]
 name = "distro"
 version = "1.9.0"