diff --git a/python/packages/core/agent_framework/_telemetry.py b/python/packages/core/agent_framework/_telemetry.py
index f7ca2ce030..ec3d55be4b 100644
--- a/python/packages/core/agent_framework/_telemetry.py
+++ b/python/packages/core/agent_framework/_telemetry.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import os
 from typing import Any, Final
@@ -60,13 +61,12 @@ def _detect_hosted_environment() -> None:
     global _hosted_env_detected
     if _hosted_env_detected:
         return
-    _hosted_env_detected = True
 
-    env_value = os.environ.get(_FOUNDRY_HOSTING_ENV_VAR)
-    if env_value is not None:
+    if (env_value := os.environ.get(_FOUNDRY_HOSTING_ENV_VAR)) is not None:
         # Env var exists — trust its value and skip the fallback.
         if env_value:
             _add_user_agent_prefix(_HOSTED_USER_AGENT_PREFIX)
+            _hosted_env_detected = True
         return
 
     # Env var not set — fall back to AgentConfig as a second layer of defense.
@@ -78,13 +78,12 @@ def _detect_hosted_environment() -> None:
             return
     except (ModuleNotFoundError, ValueError):
         return
-    try:
+    with contextlib.suppress(ImportError, AttributeError):
         from azure.ai.agentserver.core import AgentConfig  # pyright: ignore[reportMissingImports]
 
         if AgentConfig.from_env().is_hosted:
             _add_user_agent_prefix(_HOSTED_USER_AGENT_PREFIX)
-    except (ImportError, AttributeError):
-        pass
+            _hosted_env_detected = True
 
 
 def get_user_agent() -> str:
diff --git a/python/packages/core/agent_framework/foundry/__init__.py b/python/packages/core/agent_framework/foundry/__init__.py
index c1e47cd6b8..79736b5ca7 100644
--- a/python/packages/core/agent_framework/foundry/__init__.py
+++ b/python/packages/core/agent_framework/foundry/__init__.py
@@ -14,6 +14,7 @@
 _IMPORTS: dict[str, tuple[str, str]] = {
     "AnthropicFoundryClient": ("agent_framework_anthropic", "agent-framework-anthropic"),
     "FoundryAgent": ("agent_framework_foundry", "agent-framework-foundry"),
+    "FoundryAgentOptions": ("agent_framework_foundry", "agent-framework-foundry"),
     "FoundryChatClient": ("agent_framework_foundry", "agent-framework-foundry"),
     "FoundryChatOptions": ("agent_framework_foundry", "agent-framework-foundry"),
     "FoundryEmbeddingClient": ("agent_framework_foundry", "agent-framework-foundry"),
diff --git a/python/packages/devui/tests/devui/test_ui_memory_regression.py b/python/packages/devui/tests/devui/test_ui_memory_regression.py
index 7e5cd10e27..b042764f6c 100644
--- a/python/packages/devui/tests/devui/test_ui_memory_regression.py
+++ b/python/packages/devui/tests/devui/test_ui_memory_regression.py
@@ -655,7 +655,13 @@ async def test_devui_streaming_renderer_memory_is_bounded(
         )
 
         try:
-            websocket_url = await _get_devtools_websocket_url(debug_port)
+            try:
+                websocket_url = await _get_devtools_websocket_url(debug_port)
+            except RuntimeError as exc:
+                return_code = browser_process.poll()
+                if return_code is not None:
+                    pytest.skip(f"Chromium exited before DevTools became available (code {return_code}).")
+                pytest.skip(str(exc))
 
             async with websocket_connect(websocket_url, max_size=None) as websocket:
                 client = _CDPClient(websocket)
diff --git a/python/packages/foundry/agent_framework_foundry/__init__.py b/python/packages/foundry/agent_framework_foundry/__init__.py
index b70d1720f2..93953d667a 100644
--- a/python/packages/foundry/agent_framework_foundry/__init__.py
+++ b/python/packages/foundry/agent_framework_foundry/__init__.py
@@ -2,7 +2,7 @@
 
 import importlib.metadata
 
-from ._agent import FoundryAgent, RawFoundryAgent, RawFoundryAgentChatClient
+from ._agent import FoundryAgent, FoundryAgentOptions, RawFoundryAgent, RawFoundryAgentChatClient
 from ._chat_client import FoundryChatClient, FoundryChatOptions, RawFoundryChatClient
 from ._embedding_client import (
     FoundryEmbeddingClient,
@@ -25,6 +25,7 @@
 
 __all__ = [
     "FoundryAgent",
+    "FoundryAgentOptions",
     "FoundryChatClient",
     "FoundryChatOptions",
     "FoundryEmbeddingClient",
diff --git a/python/packages/foundry/agent_framework_foundry/_agent.py b/python/packages/foundry/agent_framework_foundry/_agent.py
index b473c787e5..da64b65cd9 100644
--- a/python/packages/foundry/agent_framework_foundry/_agent.py
+++ b/python/packages/foundry/agent_framework_foundry/_agent.py
@@ -16,6 +16,7 @@
 
 from agent_framework import (
     AgentMiddlewareLayer,
+    AgentSession,
     ChatAndFunctionMiddlewareTypes,
     ChatMiddlewareLayer,
     ContextProvider,
@@ -52,11 +53,13 @@
 if TYPE_CHECKING:
     from agent_framework import (
         Agent,
+        AgentRunInputs,
         ChatAndFunctionMiddlewareTypes,
         ContextProvider,
         MiddlewareTypes,
         ToolTypes,
     )
+    from agent_framework._agents import _RunContext  # pyright: ignore[reportPrivateUsage]
 
 logger: logging.Logger = logging.getLogger("agent_framework.foundry")
 
@@ -81,14 +84,54 @@ class FoundryAgentSettings(TypedDict, total=False):
     agent_version: str | None
 
 
+class FoundryAgentOptions(OpenAIChatOptions, total=False):
+    """Microsoft Foundry agent-specific chat options.
+
+    Extends ``OpenAIChatOptions`` with hosted-agent session configuration used by
+    ``FoundryAgent`` / ``RawFoundryAgent``.
+
+    Keyword Args:
+        extra_body: Additional request body values sent to the Responses API.
+        isolation_key: Isolation key used when lazily creating a hosted-agent
+            session through ``project_client.beta.agents.create_session(...)``.
+    """
+
+    extra_body: dict[str, Any]
+    isolation_key: str
+
+
 FoundryAgentOptionsT = TypeVar(
     "FoundryAgentOptionsT",
     bound=TypedDict,  # type: ignore[valid-type]
-    default="OpenAIChatOptions",
+    default="FoundryAgentOptions",
     covariant=True,
 )
 
 
+def _merge_extra_body(extra_body: Any | None, *, additions: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    """Normalize and merge provider-specific extra_body values."""
+    if extra_body is None:
+        merged: dict[str, Any] = {}
+    elif isinstance(extra_body, Mapping):
+        merged = dict(cast(Mapping[str, Any], extra_body))
+    else:
+        raise TypeError(f"extra_body must be a mapping when provided, got {type(extra_body).__name__}.")
+
+    if additions:
+        merged.update(additions)
+    return merged
+
+
+def _uses_foundry_agent_session(conversation_id: Any) -> bool:
+    """Return whether a conversation_id should be treated as a Foundry agent session id."""
+    return (
+        isinstance(conversation_id, str)
+        and bool(conversation_id)
+        and not conversation_id.startswith("resp_")
+        and not conversation_id.startswith("conv_")
+    )
+
+
 class RawFoundryAgentChatClient(  # type: ignore[misc]
     RawOpenAIChatClient[FoundryAgentOptionsT],
     Generic[FoundryAgentOptionsT],
@@ -167,13 +210,15 @@ def __init__(
         )
 
         resolved_endpoint = settings.get("project_endpoint")
-        self.agent_name = settings.get("agent_name")
-        self.agent_version = settings.get("agent_version")
+        agent_name_setting = settings.get("agent_name")
+        self.agent_version: str | None = settings.get("agent_version")
+        self.allow_preview = allow_preview or False
 
-        if not self.agent_name:
+        if not agent_name_setting:
             raise ValueError(
                 "Agent name is required. Set via 'agent_name' parameter or 'FOUNDRY_AGENT_NAME' environment variable."
             )
+        self.agent_name = agent_name_setting
 
         # Create or use provided project client
         self._should_close_client = False
@@ -197,11 +242,13 @@ def __init__(
             self.project_client = AIProjectClient(**project_client_kwargs)
             self._should_close_client = True
 
-        # Get OpenAI client from project
-        async_client = self.project_client.get_openai_client()
-
+        openai_client_kwargs: dict[str, Any] = {}
+        if default_headers:
+            openai_client_kwargs["default_headers"] = dict(default_headers)
+        if allow_preview:
+            openai_client_kwargs["agent_name"] = self.agent_name
         super().__init__(
-            async_client=async_client,
+            async_client=self.project_client.get_openai_client(**openai_client_kwargs),
             default_headers=default_headers,
             instruction_role=instruction_role,
             compaction_strategy=compaction_strategy,
@@ -209,13 +256,6 @@ def __init__(
             additional_properties=additional_properties,
         )
 
-    def _get_agent_reference(self) -> dict[str, str]:
-        """Build the agent reference dict for the Responses API."""
-        ref: dict[str, str] = {"name": self.agent_name, "type": "agent_reference"}  # type: ignore[dict-item]
-        if self.agent_version:
-            ref["version"] = self.agent_version
-        return ref
-
     @override
     def as_agent(
         self,
@@ -270,7 +310,7 @@ async def _prepare_options(
         options: Mapping[str, Any],
         **kwargs: Any,
     ) -> dict[str, Any]:
-        """Prepare options for the Responses API, injecting agent reference and validating tools."""
+        """Prepare options for the Responses API and validate client-side tools."""
         # Validate tools — only FunctionTool allowed
         tools = options.get("tools", [])
         if tools:
@@ -292,18 +332,58 @@ async def _prepare_options(
         if "input" in run_options and isinstance(run_options["input"], list):
             run_options["input"] = self._transform_input_for_azure_ai(cast(list[dict[str, Any]], run_options["input"]))
 
-        # Inject agent reference
-        run_options["extra_body"] = {"agent_reference": self._get_agent_reference()}
+        # Merge caller-supplied extra_body with any agent-specific request payload.
+        conversation_id = options.get("conversation_id")
+        extra_body = _merge_extra_body(run_options.pop("extra_body", None))
+        if _uses_foundry_agent_session(conversation_id):
+            run_options.pop("previous_response_id", None)
+            run_options.pop("conversation", None)
+            extra_body["agent_session_id"] = conversation_id
+        if extra_body:
+            run_options["extra_body"] = extra_body
+
+        run_options.pop("isolation_key", None)
 
         # Strip tools from request body - Foundry API rejects requests with both
-        # agent_reference and tools present. FunctionTools are invoked client-side
+        # agent endpoint and tools present. FunctionTools are invoked client-side
         # by the function invocation layer, not sent to the service.
-        run_options.pop("tools", None)
-        run_options.pop("tool_choice", None)
-        run_options.pop("parallel_tool_calls", None)
+        run_options.pop("model", None)
+        if not self.allow_preview:
+            run_options.pop("tools", None)
+            run_options.pop("tool_choice", None)
+            run_options.pop("parallel_tool_calls", None)
 
         return run_options
 
+    @override
+    def _parse_response_from_openai(
+        self,
+        response: Any,
+        options: dict[str, Any],
+    ) -> Any:
+        parsed_response = super()._parse_response_from_openai(response, options)
+        if _uses_foundry_agent_session(options.get("conversation_id")):
+            parsed_response.conversation_id = None
+        return parsed_response
+
+    @override
+    def _parse_chunk_from_openai(
+        self,
+        event: Any,
+        options: dict[str, Any],
+        function_call_ids: dict[int, tuple[str, str]],
+        seen_reasoning_delta_item_ids: set[str] | None = None,
+    ) -> Any:
+        parsed_chunk = super()._parse_chunk_from_openai(
+            event,
+            options,
+            function_call_ids,
+            seen_reasoning_delta_item_ids,
+        )
+        if _uses_foundry_agent_session(options.get("conversation_id")):
+            parsed_chunk.conversation_id = None
+        return parsed_chunk
+
     @override
     def _check_model_presence(self, options: dict[str, Any]) -> None:
         """Skip model check — model is configured on the Foundry agent."""
@@ -368,6 +448,26 @@ def _transform_input_for_azure_ai(self, input_items: list[dict[str, Any]]) -> li
 
         return transformed
 
+    async def get_agent_version(self) -> str | None:
+        """Return the agent version if available, else None."""
+        if self.agent_version is not None:
+            return self.agent_version
+        if not self.allow_preview:
+            return None
+        agent_details = await cast(Any, self.project_client.beta.agents).get(  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
+            agent_name=self.agent_name
+        )
+        versions_object = getattr(agent_details, "versions", None)
+        if not isinstance(versions_object, Mapping):
+            raise TypeError("Foundry agent details did not include a versions mapping.")
+        versions = cast(Mapping[str, Any], versions_object)
+        latest_version = versions.get("latest")
+        agent_version = getattr(cast(Any, latest_version), "version", None)
+        if not isinstance(agent_version, str):
+            raise TypeError("Foundry agent details did not include a latest version string.")
+        self.agent_version = agent_version
+        return agent_version
+
     async def close(self) -> None:
         """Close the project client if we created it."""
         if self._should_close_client:
@@ -395,7 +495,7 @@ class _FoundryAgentChatClient(  # type: ignore[misc]
             client = FoundryAgentClient(
                 project_endpoint="https://your-project.services.ai.azure.com",
                 agent_name="my-prompt-agent",
-                agent_version="1.0",
+                agent_version="1",
                 credential=AzureCliCredential(),
             )
 
@@ -477,7 +577,7 @@ class RawFoundryAgent(  # type: ignore[misc]
             agent = RawFoundryAgent(
                 project_endpoint="https://your-project.services.ai.azure.com",
                 agent_name="my-prompt-agent",
-                agent_version="1.0",
+                agent_version="1",
                 credential=AzureCliCredential(),
             )
             result = await agent.run("Hello!")
@@ -570,7 +670,7 @@ def __init__(
             client=client,  # type: ignore[arg-type]
             instructions=instructions,
             id=id,
-            name=name,
+            name=name or agent_name,
             description=description,
             tools=tools,  # type: ignore[arg-type]
             default_options=cast(FoundryAgentOptionsT | None, default_options),
@@ -582,6 +682,81 @@ def __init__(
             additional_properties=dict(additional_properties) if additional_properties is not None else None,
         )
 
+    def _resolve_service_session_isolation_key(self, isolation_key: str | None = None) -> str:
+        """Resolve the isolation key from an explicit value or default_options."""
+        resolved_isolation_key = (
+            isolation_key if isolation_key is not None else self.default_options.get("isolation_key")
+        )
+        if resolved_isolation_key is None:
+            raise ValueError("isolation_key is required. Pass it explicitly or set default_options['isolation_key'].")
+        return resolved_isolation_key
+
+    async def _create_service_session_id(
+        self,
+        *,
+        isolation_key: str | None = None,
+    ) -> str:
+        """Create a hosted Foundry service session and return the service session ID."""
+        if not isinstance(self.client, RawFoundryAgentChatClient):
+            raise TypeError("_create_service_session_id requires a RawFoundryAgentChatClient-based client.")
+        if not self.client.allow_preview:
+            raise RuntimeError("Hosted Foundry service sessions require allow_preview=True.")
+
+        create_session_kwargs: dict[str, Any] = {
+            "agent_name": self.client.agent_name,
+            "isolation_key": self._resolve_service_session_isolation_key(isolation_key),
+        }
+        if version := await self.client.get_agent_version():
+            from azure.ai.projects.models import VersionRefIndicator
+
+            create_session_kwargs["version_indicator"] = VersionRefIndicator(agent_version=version)  # type: ignore
+
+        service_session = await self.client.project_client.beta.agents.create_session(**create_session_kwargs)
+        agent_session_id = getattr(service_session, "agent_session_id", None)
+        if not isinstance(agent_session_id, str) or not agent_session_id:
+            raise ValueError("Hosted Foundry session creation did not return a non-empty agent_session_id.")
+
+        return agent_session_id
+
+    @override
+    async def _prepare_run_context(
+        self,
+        *,
+        messages: AgentRunInputs | None,
+        session: AgentSession | None,
+        tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None,
+        options: Mapping[str, Any] | None,
+        compaction_strategy: CompactionStrategy | None,
+        tokenizer: TokenizerProtocol | None,
+        function_invocation_kwargs: Mapping[str, Any] | None,
+        client_kwargs: Mapping[str, Any] | None,
+    ) -> _RunContext:
+        runtime_options = dict(options) if options else {}
+        effective_options = {
+            **{key: value for key, value in self.default_options.items() if value is not None},
+            **{key: value for key, value in runtime_options.items() if value is not None},
+        }
+
+        if (
+            session is not None
+            and session.service_session_id is None
+            and effective_options.get("isolation_key") is not None
+        ):
+            session.service_session_id = await self._create_service_session_id(
+                isolation_key=cast(str | None, effective_options.get("isolation_key")),
+            )
+
+        return await super()._prepare_run_context(
+            messages=messages,
+            session=session,
+            tools=tools,
+            options=runtime_options,
+            compaction_strategy=compaction_strategy,
+            tokenizer=tokenizer,
+            function_invocation_kwargs=function_invocation_kwargs,
+            client_kwargs=client_kwargs,
+        )
+
     async def configure_azure_monitor(
         self,
         enable_sensitive_data: bool = False,
@@ -708,6 +883,19 @@ def __init__(
     ) -> None:
         """Initialize a Foundry Agent with full middleware and telemetry.
 
+        ``FoundryAgent`` supports both PromptAgents and HostedAgents. PromptAgents
+        typically provide ``agent_version`` directly. HostedAgents can omit
+        ``agent_version`` and, when they need preview-only session APIs, should
+        opt in with ``allow_preview=True`` when this class creates the underlying
+        ``AIProjectClient``. If you pass ``project_client`` explicitly, it must
+        already be configured for preview APIs before being passed to
+        ``FoundryAgent``.
+
+        To lazily create HostedAgent service sessions inside the agent, pass an
+        ``isolation_key`` through ``default_options`` (or per-run options). The
+        agent stores the resulting HostedAgent session ID in
+        ``AgentSession.service_session_id`` and reuses it on subsequent runs.
+
         Keyword Args:
             project_endpoint: The Foundry project endpoint URL.
             agent_name: The name of the Foundry agent to connect to.
@@ -715,6 +903,9 @@ def __init__(
             credential: Azure credential for authentication.
             project_client: An existing AIProjectClient to use.
             allow_preview: Enables preview opt-in on internally-created AIProjectClient.
+                Set this to ``True`` for HostedAgents that need preview-only
+                session APIs, including lazy service session creation from
+                ``isolation_key``.
             tools: Function tools to provide to the agent. Only ``FunctionTool`` objects are accepted.
             context_providers: Optional context providers.
             middleware: Optional agent-level middleware.
@@ -726,6 +917,8 @@ def __init__(
             description: Optional local description for the local agent wrapper.
             instructions: Optional instructions for the local agent wrapper.
             default_options: Default chat options for the local agent wrapper.
+                ``FoundryAgentOptions`` can include ``isolation_key`` and
+                ``extra_body`` when working with HostedAgents.
             require_per_service_call_history_persistence: Whether to require per-service-call
                 chat history persistence when using local history providers.
             function_invocation_configuration: Optional function invocation configuration override.
diff --git a/python/packages/foundry/agent_framework_foundry/_chat_client.py b/python/packages/foundry/agent_framework_foundry/_chat_client.py
index 4428d69dc6..57522fb886 100644
--- a/python/packages/foundry/agent_framework_foundry/_chat_client.py
+++ b/python/packages/foundry/agent_framework_foundry/_chat_client.py
@@ -204,9 +204,13 @@ def __init__(
                 project_client_kwargs["allow_preview"] = allow_preview
             project_client = AIProjectClient(**project_client_kwargs)
 
+        openai_kwargs: dict[str, Any] = {}
+        if default_headers:
+            openai_kwargs["default_headers"] = default_headers
+
         super().__init__(
             model=resolved_model,
-            async_client=project_client.get_openai_client(),
+            async_client=project_client.get_openai_client(**openai_kwargs),
             default_headers=default_headers,
             instruction_role=instruction_role,
             compaction_strategy=compaction_strategy,
diff --git a/python/packages/foundry/tests/foundry/test_foundry_agent.py b/python/packages/foundry/tests/foundry/test_foundry_agent.py
index 829af6ab87..73670d0bbc 100644
--- a/python/packages/foundry/tests/foundry/test_foundry_agent.py
+++ b/python/packages/foundry/tests/foundry/test_foundry_agent.py
@@ -5,11 +5,12 @@
 import inspect
 import os
 import sys
+from types import SimpleNamespace
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from agent_framework import AgentResponse, ChatContext, ChatMiddleware, Message, tool
+from agent_framework import AgentResponse, AgentSession, ChatContext, ChatMiddleware, ChatResponse, Message, tool
 from azure.core.exceptions import ResourceNotFoundError
 from azure.identity import AzureCliCredential
 
@@ -54,7 +55,7 @@ def test_raw_foundry_agent_chat_client_init_requires_agent_name() -> None:
 
 
 def test_raw_foundry_agent_chat_client_init_with_agent_name() -> None:
-    """Test construction with agent_name and project_client."""
+    """Test construction with agent_name and project_client without preview agent binding."""
 
     mock_project = MagicMock()
     mock_project.get_openai_client.return_value = MagicMock()
@@ -67,49 +68,38 @@ def test_raw_foundry_agent_chat_client_init_with_agent_name() -> None:
 
     assert client.agent_name == "test-agent"
     assert client.agent_version == "1.0"
+    mock_project.get_openai_client.assert_called_once_with()
 
 
-def test_raw_foundry_agent_chat_client_init_uses_explicit_parameters() -> None:
-    signature = inspect.signature(RawFoundryAgentChatClient.__init__)
-
-    assert "default_headers" in signature.parameters
-    assert "instruction_role" in signature.parameters
-    assert "compaction_strategy" in signature.parameters
-    assert "tokenizer" in signature.parameters
-    assert "additional_properties" in signature.parameters
-    assert all(parameter.kind != inspect.Parameter.VAR_KEYWORD for parameter in signature.parameters.values())
-
-
-def test_raw_foundry_agent_chat_client_get_agent_reference_with_version() -> None:
-    """Test agent reference includes version when provided."""
+def test_raw_foundry_agent_chat_client_init_passes_agent_name_when_preview_enabled() -> None:
+    """Test preview-enabled clients bind the OpenAI client to the agent endpoint."""
 
     mock_project = MagicMock()
     mock_project.get_openai_client.return_value = MagicMock()
 
     client = RawFoundryAgentChatClient(
         project_client=mock_project,
-        agent_name="my-agent",
-        agent_version="2.0",
+        agent_name="hosted-agent",
+        allow_preview=True,
+        default_headers={"x-test": "1"},
     )
 
-    ref = client._get_agent_reference()
-    assert ref == {"name": "my-agent", "version": "2.0", "type": "agent_reference"}
-
-
-def test_raw_foundry_agent_chat_client_get_agent_reference_without_version() -> None:
-    """Test agent reference omits version for HostedAgents."""
-
-    mock_project = MagicMock()
-    mock_project.get_openai_client.return_value = MagicMock()
-
-    client = RawFoundryAgentChatClient(
-        project_client=mock_project,
+    assert client.agent_name == "hosted-agent"
+    mock_project.get_openai_client.assert_called_once_with(
         agent_name="hosted-agent",
+        default_headers={"x-test": "1"},
     )
 
-    ref = client._get_agent_reference()
-    assert ref == {"name": "hosted-agent", "type": "agent_reference"}
-    assert "version" not in ref
+
+def test_raw_foundry_agent_chat_client_init_uses_explicit_parameters() -> None:
+    signature = inspect.signature(RawFoundryAgentChatClient.__init__)
+
+    assert "default_headers" in signature.parameters
+    assert "instruction_role" in signature.parameters
+    assert "compaction_strategy" in signature.parameters
+    assert "tokenizer" in signature.parameters
+    assert "additional_properties" in signature.parameters
+    assert all(parameter.kind != inspect.Parameter.VAR_KEYWORD for parameter in signature.parameters.values())
 
 
 def test_raw_foundry_agent_chat_client_as_agent_preserves_client_type() -> None:
@@ -196,12 +186,11 @@ def my_func() -> str:
             options={"tools": [my_func]},
         )
 
-    assert "extra_body" in result
-    assert result["extra_body"]["agent_reference"]["name"] == "test-agent"
+    assert result == {}
 
 
-async def test_raw_foundry_agent_chat_client_prepare_options_strips_tools() -> None:
-    """Test that _prepare_options strips tools, tool_choice, and parallel_tool_calls from run_options."""
+async def test_raw_foundry_agent_chat_client_prepare_options_strips_client_side_fields() -> None:
+    """Test that _prepare_options strips model and tool-loop fields from run_options."""
 
     mock_project = MagicMock()
     mock_openai = MagicMock()
@@ -222,6 +211,7 @@ def my_func() -> str:
         "agent_framework_openai._chat_client.RawOpenAIChatClient._prepare_options",
         new_callable=AsyncMock,
         return_value={
+            "model": "gpt-4.1",
             "tools": [{"type": "function", "function": {"name": "my_func"}}],
             "tool_choice": "auto",
             "parallel_tool_calls": True,
@@ -232,11 +222,69 @@ def my_func() -> str:
             options={"tools": [my_func]},
         )
 
+    assert "model" not in result
     assert "tools" not in result
     assert "tool_choice" not in result
     assert "parallel_tool_calls" not in result
-    assert "extra_body" in result
-    assert result["extra_body"]["agent_reference"]["name"] == "test-agent"
+    assert result == {}
+
+
+async def test_raw_foundry_agent_chat_client_prepare_options_maps_agent_session_id_to_extra_body() -> None:
+    """Test that service_session_id is forwarded as agent_session_id for hosted sessions."""
+
+    mock_project = MagicMock()
+    mock_openai = MagicMock()
+    mock_project.get_openai_client.return_value = mock_openai
+
+    client = RawFoundryAgentChatClient(
+        project_client=mock_project,
+        agent_name="test-agent",
+    )
+
+    with patch(
+        "agent_framework_openai._chat_client.RawOpenAIChatClient._prepare_options",
+        new_callable=AsyncMock,
+        return_value={
+            "extra_body": {"custom": "value"},
+            "previous_response_id": "should-be-removed",
+        },
+    ):
+        result = await client._prepare_options(
+            messages=[Message(role="user", contents="hi")],
+            options={"conversation_id": "agent-session-123", "isolation_key": "iso-key"},
+        )
+
+    assert result["extra_body"] == {
+        "custom": "value",
+        "agent_session_id": "agent-session-123",
+    }
+    assert "previous_response_id" not in result
+    assert "conversation" not in result
+    assert "isolation_key" not in result
+
+
+def test_raw_foundry_agent_chat_client_parse_response_suppresses_conversation_id_for_agent_sessions() -> None:
+    """Test that agent-session continuations do not overwrite session.service_session_id."""
+
+    mock_project = MagicMock()
+    mock_project.get_openai_client.return_value = MagicMock()
+
+    client = RawFoundryAgentChatClient(
+        project_client=mock_project,
+        agent_name="test-agent",
+    )
+
+    parsed = ChatResponse(conversation_id="resp_123")
+    with patch(
+        "agent_framework_openai._chat_client.RawOpenAIChatClient._parse_response_from_openai",
+        return_value=parsed,
+    ):
+        result = client._parse_response_from_openai(
+            response=MagicMock(),
+            options={"conversation_id": "agent-session-123"},
+        )
+
+    assert result.conversation_id is None
 
 
 def test_raw_foundry_agent_chat_client_check_model_presence_is_noop() -> None:
@@ -366,6 +414,74 @@ def my_func() -> str:
     assert agent.default_options.get("tools") is not None
 
 
+async def test_raw_foundry_agent_prepare_run_context_creates_service_session_from_isolation_key() -> None:
+    """Test that RawFoundryAgent lazily creates a hosted session and stores it on service_session_id."""
+
+    mock_project = MagicMock()
+    mock_project.get_openai_client.return_value = MagicMock()
+    mock_project.beta = SimpleNamespace(
+        agents=SimpleNamespace(
+            create_session=AsyncMock(return_value=SimpleNamespace(agent_session_id="agent-session-123"))
+        )
+    )
+
+    agent = RawFoundryAgent(
+        project_client=mock_project,
+        agent_name="test-agent",
+        agent_version="1.0",
+        allow_preview=True,
+    )
+    session = AgentSession()
+
+    with patch(
+        "agent_framework._agents.RawAgent._prepare_run_context",
+        new=AsyncMock(return_value={"ok": True}),
+    ) as mock_prepare_run_context:
+        result = await agent._prepare_run_context(
+            messages="hi",
+            session=session,
+            tools=None,
+            options={"isolation_key": "iso-key"},
+            compaction_strategy=None,
+            tokenizer=None,
+            function_invocation_kwargs=None,
+            client_kwargs=None,
+        )
+
+    assert result == {"ok": True}
+    assert session.service_session_id == "agent-session-123"
+    mock_project.beta.agents.create_session.assert_awaited_once()
+    create_session_kwargs = mock_project.beta.agents.create_session.await_args.kwargs
+    assert create_session_kwargs["agent_name"] == "test-agent"
+    assert create_session_kwargs["isolation_key"] == "iso-key"
+    assert "version_indicator" in create_session_kwargs
+    mock_prepare_run_context.assert_awaited_once()
+
+
+async def test_raw_foundry_agent_prepare_run_context_requires_preview_for_hosted_sessions() -> None:
+    """Test that hosted-agent sessions require allow_preview=True."""
+
+    mock_project = MagicMock()
+    mock_project.get_openai_client.return_value = MagicMock()
+
+    agent = RawFoundryAgent(
+        project_client=mock_project,
+        agent_name="test-agent",
+    )
+
+    with pytest.raises(RuntimeError, match="allow_preview=True"):
+        await agent._prepare_run_context(
+            messages="hi",
+            session=AgentSession(),
+            tools=None,
+            options={"isolation_key": "iso-key"},
+            compaction_strategy=None,
+            tokenizer=None,
+            function_invocation_kwargs=None,
+            client_kwargs=None,
+        )
+
+
 def test_foundry_agent_init() -> None:
     """Test construction of the full-middleware agent."""
 
@@ -483,9 +599,10 @@ def _import_with_missing_azure_monitor(
 @pytest.mark.flaky
 @pytest.mark.integration
 @skip_if_foundry_agent_integration_tests_disabled
+@pytest.mark.skip(reason="Test agent seems to have disappeared from the test environment; needs investigation.")
 async def test_foundry_agent_basic_run() -> None:
     """Smoke-test FoundryAgent against a real configured agent."""
-    async with FoundryAgent(credential=AzureCliCredential()) as agent:
+    async with FoundryAgent(credential=AzureCliCredential(), allow_preview=True) as agent:
         response = await agent.run("Please respond with exactly: 'This is a response test.'")
 
     assert isinstance(response, AgentResponse)
@@ -496,6 +613,7 @@ async def test_foundry_agent_basic_run() -> None:
 @pytest.mark.flaky
 @pytest.mark.integration
 @skip_if_foundry_agent_integration_tests_disabled
+@pytest.mark.skip(reason="Test agent seems to have disappeared from the test environment; needs investigation.")
 async def test_foundry_agent_custom_client_run() -> None:
     """Smoke-test FoundryAgent against a real configured agent."""
     async with FoundryAgent(credential=AzureCliCredential(), client_type=RawFoundryAgentChatClient) as agent:
diff --git a/python/packages/foundry/tests/foundry/test_foundry_embedding_client.py b/python/packages/foundry/tests/foundry/test_foundry_embedding_client.py
index e9e342d675..664123637d 100644
--- a/python/packages/foundry/tests/foundry/test_foundry_embedding_client.py
+++ b/python/packages/foundry/tests/foundry/test_foundry_embedding_client.py
@@ -198,6 +198,7 @@ def test_settings_from_env(self) -> None:
                     "FOUNDRY_MODELS_API_KEY": "env-key",
                     "FOUNDRY_EMBEDDING_MODEL": "env-model",
                 },
+                clear=True,
             ),
             patch("agent_framework_foundry._embedding_client.EmbeddingsClient"),
             patch("agent_framework_foundry._embedding_client.ImageEmbeddingsClient"),
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index cac0ac3790..9078c59d22 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -172,12 +172,7 @@ def __init__(
         self._agent = agent
         self.response_handler(self._handle_response)  # pyright: ignore[reportUnknownMemberType]
 
-    @staticmethod
-    def _is_streaming_request(request: CreateResponse) -> bool:
-        """Check if the request is a streaming request."""
-        return request.stream is not None and request.stream is True
-
-    def _handle_response(
+    async def _handle_response(
         self,
         request: CreateResponse,
         context: ResponseContext,
@@ -186,11 +181,10 @@ def _handle_response(
         """Handle the creation of a response."""
         if self._is_workflow_agent:
             # Workflow agents are handled differently because they require checkpoint restoration
-            return self._handle_workflow_agent(request, context)
-
-        return self._handle_regular_agent(request, context)
+            return self._handle_inner_workflow(request, context)
+        return self._handle_inner_agent(request, context)
 
-    async def _handle_regular_agent(
+    async def _handle_inner_agent(
         self,
         request: CreateResponse,
         context: ResponseContext,
@@ -200,25 +194,24 @@ async def _handle_regular_agent(
         input_messages = _items_to_messages(input_items)
 
         history = await context.get_history()
-        messages: list[str | Content | Message] = [*_output_items_to_messages(history), *input_messages]
+        run_kwargs: dict[str, Any] = {"messages": [*_output_items_to_messages(history), *input_messages]}
+        is_streaming_request = request.stream is not None and request.stream is True
 
         chat_options, are_options_set = _to_chat_options(request)
 
-        is_streaming_request = self._is_streaming_request(request)
         response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model)
 
         yield response_event_stream.emit_created()
         yield response_event_stream.emit_in_progress()
 
+        if are_options_set and not isinstance(self._agent, RawAgent):
+            logger.warning("Agent doesn't support runtime options. They will be ignored.")
+        else:
+            run_kwargs["options"] = chat_options
+
         if not is_streaming_request:
             # Run the agent in non-streaming mode
-            if isinstance(self._agent, RawAgent):
-                raw_agent = cast("RawAgent[Any]", self._agent)  # type: ignore[redundant-cast]  # pyright: ignore[reportUnknownMemberType]
-                response = await raw_agent.run(messages, stream=False, options=chat_options)
-            else:
-                if are_options_set:
-                    logger.warning("Agent doesn't support runtime options. They will be ignored.")
-                response = await self._agent.run(messages, stream=False)
+            response = await self._agent.run(stream=False, **run_kwargs)  # type: ignore[reportUnknownMemberType]
 
             for message in response.messages:
                 for content in message.contents:
@@ -228,20 +221,12 @@ async def _handle_regular_agent(
             yield response_event_stream.emit_completed()
             return
 
-        # Run the agent in streaming mode
-        if isinstance(self._agent, RawAgent):
-            raw_agent = cast("RawAgent[Any]", self._agent)  # type: ignore[redundant-cast]  # pyright: ignore[reportUnknownMemberType]
-            response_stream = raw_agent.run(messages, stream=True, options=chat_options)
-        else:
-            if are_options_set:
-                logger.warning("Agent doesn't support runtime options. They will be ignored.")
-            response_stream = self._agent.run(messages, stream=True)
-
         # Track the current active output item builder for streaming;
         # lazily created on matching content, closed when a different type arrives.
         tracker = _OutputItemTracker(response_event_stream)
 
-        async for update in response_stream:
+        # Run the agent in streaming mode
+        async for update in self._agent.run(stream=True, **run_kwargs):  # type: ignore[reportUnknownMemberType]
             for content in update.contents:
                 for event in tracker.handle(content):
                     yield event
@@ -256,7 +241,7 @@ async def _handle_regular_agent(
 
         yield response_event_stream.emit_completed()
 
-    async def _handle_workflow_agent(
+    async def _handle_inner_workflow(
         self,
         request: CreateResponse,
         context: ResponseContext,
@@ -269,8 +254,7 @@ async def _handle_workflow_agent(
         """
         input_items = await context.get_input_items()
         input_messages = _items_to_messages(input_items)
-
-        is_streaming_request = self._is_streaming_request(request)
+        is_streaming_request = request.stream is not None and request.stream is True
 
         _, are_options_set = _to_chat_options(request)
         if are_options_set:
@@ -311,7 +295,8 @@ async def _handle_workflow_agent(
         response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model)
 
         # Create a new checkpoint storage for this response based on the following rules:
-        # - If no previous response ID or conversation ID is provided, create a new checkpoint storage for this response
+        # - If no previous response ID or conversation ID is provided,
+        #   create a new checkpoint storage for this response
         # - If a previous response ID is provided, create a new checkpoint storage for this response
         # - If a conversation ID is provided, reuse the existing checkpoint storage for the conversation
         context_id = context.conversation_id or context.response_id
@@ -333,14 +318,12 @@ async def _handle_workflow_agent(
             yield response_event_stream.emit_completed()
             return
 
-        # Run the agent in streaming mode
-        response_stream = self._agent.run(input_messages, stream=True, checkpoint_storage=checkpoint_storage)
-
         # Track the current active output item builder for streaming;
         # lazily created on matching content, closed when a different type arrives.
         tracker = _OutputItemTracker(response_event_stream)
 
-        async for update in response_stream:
+        # Run the workflow agent in streaming mode
+        async for update in self._agent.run(input_messages, stream=True, checkpoint_storage=checkpoint_storage):
             for content in update.contents:
                 for event in tracker.handle(content):
                     yield event
@@ -355,7 +338,6 @@ async def _handle_workflow_agent(
 
         await self._delete_not_latest_checkpoints(checkpoint_storage, self._agent.workflow.name)
         yield response_event_stream.emit_completed()
-        return
 
     @staticmethod
     async def _delete_not_latest_checkpoints(checkpoint_storage: FileCheckpointStorage, workflow_name: str) -> None:
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 13538b6c9a..237a3c7634 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -41,9 +41,10 @@ def _make_agent(
     *,
     response: AgentResponse | None = None,
     stream_updates: list[AgentResponseUpdate] | None = None,
+    raw_agent: bool = True,
 ) -> MagicMock:
     """Create a mock agent implementing SupportsAgentRun."""
-    agent = MagicMock(spec=RawAgent)
+    agent = MagicMock(spec=RawAgent) if raw_agent else MagicMock()
     agent.id = "test-agent"
     agent.name = "Test Agent"
     agent.description = "A mock agent for testing"
@@ -267,10 +268,18 @@ async def test_empty_response(self) -> None:
 
     async def test_chat_options_forwarded(self) -> None:
         agent = _make_agent(
-            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])]),
+            raw_agent=True,
         )
         server = _make_server(agent)
-        resp = await _post(server, stream=False, temperature=0.5, top_p=0.9, max_output_tokens=1024)
+        resp = await _post(
+            server,
+            stream=False,
+            temperature=0.5,
+            top_p=0.9,
+            max_output_tokens=1024,
+            parallel_tool_calls=True,
+        )
 
         assert resp.status_code == 200
         agent.run.assert_awaited_once()
@@ -280,6 +289,7 @@ async def test_chat_options_forwarded(self) -> None:
         assert options["temperature"] == 0.5
         assert options["top_p"] == 0.9
         assert options["max_tokens"] == 1024
+        assert options["allow_multiple_tool_calls"] is True
 
 
 # endregion
@@ -289,6 +299,31 @@ async def test_chat_options_forwarded(self) -> None:
 
 
 class TestStreaming:
+    async def test_chat_options_forwarded(self) -> None:
+        agent = _make_agent(
+            stream_updates=[AgentResponseUpdate(contents=[Content.from_text("ok")], role="assistant")],
+            raw_agent=True,
+        )
+        server = _make_server(agent)
+        resp = await _post(
+            server,
+            stream=True,
+            temperature=0.5,
+            top_p=0.9,
+            max_output_tokens=1024,
+            parallel_tool_calls=True,
+        )
+
+        assert resp.status_code == 200
+        agent.run.assert_called_once()
+        call_kwargs = agent.run.call_args.kwargs
+        assert call_kwargs["stream"] is True
+        options = call_kwargs["options"]
+        assert options["temperature"] == 0.5
+        assert options["top_p"] == 0.9
+        assert options["max_tokens"] == 1024
+        assert options["allow_multiple_tool_calls"] is True
+
     async def test_basic_text_streaming(self) -> None:
         agent = _make_agent(
             stream_updates=[
@@ -1426,7 +1461,7 @@ async def test_text_and_image_input_single_turn(self) -> None:
         assert body["status"] == "completed"
 
         # Verify agent received text + image
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 1
         assert messages[0].role == "user"
         assert len(messages[0].contents) == 2
@@ -1464,7 +1499,7 @@ async def test_text_and_file_input_single_turn(self) -> None:
         body = resp.json()
         assert body["status"] == "completed"
 
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 1
         assert len(messages[0].contents) == 2
         assert messages[0].contents[0].type == "text"
@@ -1501,7 +1536,7 @@ async def test_mixed_text_and_image_input(self) -> None:
         body = resp.json()
         assert body["status"] == "completed"
 
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 1
         assert len(messages[0].contents) == 2
         assert messages[0].contents[0].type == "text"
@@ -1542,7 +1577,7 @@ async def test_function_call_items_in_input(self) -> None:
         body = resp.json()
         assert body["status"] == "completed"
 
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 3
         assert messages[0].role == "user"
         assert messages[0].contents[0].type == "text"
@@ -1591,7 +1626,7 @@ async def test_multi_turn_text_then_text_with_image(self) -> None:
         assert body2["status"] == "completed"
 
         # Verify second call receives history from turn 1 + text+image input
-        second_call_messages = agent.run.call_args_list[1].args[0]
+        second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
         # History: output message from turn 1 ("Send me an image")
         # Input: message with text + image
         assert len(second_call_messages) >= 2
@@ -1652,7 +1687,7 @@ async def test_multi_turn_function_call_in_history(self) -> None:
         assert resp2.json()["status"] == "completed"
 
         # Verify turn 2 received history including function call/result
-        second_call_messages = agent.run.call_args_list[1].args[0]
+        second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
         roles = [m.role for m in second_call_messages]
         assert "assistant" in roles
         assert "tool" in roles
@@ -1703,7 +1738,7 @@ async def test_multi_turn_reasoning_in_history(self) -> None:
         assert resp2.json()["status"] == "completed"
 
         # Verify history includes the reasoning and text from turn 1
-        second_call_messages = agent.run.call_args_list[1].args[0]
+        second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
         assert len(second_call_messages) >= 2  # history + new input
 
     async def test_multi_turn_with_mixed_content_and_streaming(self) -> None:
@@ -1795,7 +1830,7 @@ async def test_text_with_mcp_call_items(self) -> None:
         body = resp.json()
         assert body["status"] == "completed"
 
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 2
         assert messages[0].role == "user"
         assert messages[0].contents[0].type == "text"
@@ -1867,7 +1902,7 @@ async def test_three_turn_conversation_with_mixed_content(self) -> None:
         assert resp3.json()["status"] == "completed"
 
         # Verify turn 3 received full history from turns 1+2 plus new image input
-        third_call_messages = agent.run.call_args_list[2].args[0]
+        third_call_messages = agent.run.call_args_list[2].kwargs["messages"]
         # Should have: history from turn 1 (assistant text) + history from turn 2
         # (function_call, function_call_output, text) + new input (text + image)
         assert len(third_call_messages) >= 5
@@ -1918,7 +1953,7 @@ async def test_input_with_hosted_file_image(self) -> None:
         body = resp.json()
         assert body["status"] == "completed"
 
-        messages = agent.run.call_args.args[0]
+        messages = agent.run.call_args.kwargs["messages"]
         assert len(messages) == 1
         assert len(messages[0].contents) == 2
         assert messages[0].contents[0].type == "text"
@@ -1982,7 +2017,7 @@ async def test_multi_turn_text_and_image_then_text_and_file(self) -> None:
         assert resp2.json()["status"] == "completed"
 
         # Verify turn 2 received history from turn 1 + new text+file input
-        second_call_messages = agent.run.call_args_list[1].args[0]
+        second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
         assert len(second_call_messages) >= 2
 
         # History should include the assistant response from turn 1
@@ -2050,7 +2085,7 @@ async def test_multi_turn_function_call_then_text_and_image(self) -> None:
         assert resp2.json()["status"] == "completed"
 
         # Verify turn 2 received history with function call + new text+image
-        second_call_messages = agent.run.call_args_list[1].args[0]
+        second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
         # History should contain function_call and function_result from turn 1
         fc_contents = [
             c for m in second_call_messages if m.role == "assistant" for c in m.contents if c.type == "function_call"
diff --git a/python/packages/gemini/tests/test_gemini_client.py b/python/packages/gemini/tests/test_gemini_client.py
index d5fcf5dbe0..480525ea1d 100644
--- a/python/packages/gemini/tests/test_gemini_client.py
+++ b/python/packages/gemini/tests/test_gemini_client.py
@@ -285,8 +285,10 @@ def test_vertex_ai_requires_project_and_location_together(monkeypatch: pytest.Mo
         GeminiChatClient(model="gemini-2.5-flash")
 
 
-async def test_missing_model_raises_on_get_response() -> None:
+async def test_missing_model_raises_on_get_response(monkeypatch: pytest.MonkeyPatch) -> None:
     """Raises ValueError at call time when no model is set on the client or in options."""
+    monkeypatch.delenv("GEMINI_MODEL", raising=False)
+    monkeypatch.delenv("GOOGLE_MODEL", raising=False)
     client, mock = _make_gemini_client(model=None)  # type: ignore[arg-type]
     mock.aio.models.generate_content = AsyncMock()
 
diff --git a/python/packages/openai/tests/openai/test_openai_chat_client_azure.py b/python/packages/openai/tests/openai/test_openai_chat_client_azure.py
index b16fbd0f7f..a5fdff72b5 100644
--- a/python/packages/openai/tests/openai/test_openai_chat_client_azure.py
+++ b/python/packages/openai/tests/openai/test_openai_chat_client_azure.py
@@ -355,6 +355,7 @@ async def test_integration_web_search() -> None:
 @pytest.mark.integration
 @skip_if_azure_openai_integration_tests_disabled
 @_with_azure_openai_debug()
+@pytest.mark.skip(reason="Azure OpenAI with files raises 500 error. Needs investigation.")
 async def test_integration_client_file_search() -> None:
     async with AzureCliCredential() as credential:
         client = OpenAIChatClient(credential=credential)
@@ -380,6 +381,7 @@ async def test_integration_client_file_search() -> None:
 @pytest.mark.integration
 @skip_if_azure_openai_integration_tests_disabled
 @_with_azure_openai_debug()
+@pytest.mark.skip(reason="Azure OpenAI with files raises 500 error. Needs investigation.")
 async def test_integration_client_file_search_streaming() -> None:
     async with AzureCliCredential() as credential:
         client = OpenAIChatClient(credential=credential)
diff --git a/python/samples/02-agents/chat_client/built_in_chat_clients.py b/python/samples/02-agents/chat_client/built_in_chat_clients.py
index 4d79cc17b4..32f3efcf57 100644
--- a/python/samples/02-agents/chat_client/built_in_chat_clients.py
+++ b/python/samples/02-agents/chat_client/built_in_chat_clients.py
@@ -75,11 +75,7 @@ def get_client(client_name: ClientName) -> SupportsChatGetResponse[Any]:
     if client_name == "azure_openai_chat_completion":
         return OpenAIChatCompletionClient(credential=AzureCliCredential())
     if client_name == "foundry_chat":
-        return FoundryChatClient(
-            project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
-            model=os.environ["FOUNDRY_MODEL"],
-            credential=AzureCliCredential(),
-        )
+        return FoundryChatClient(credential=AzureCliCredential())
 
     raise ValueError(f"Unsupported client name: {client_name}")
 
@@ -93,21 +89,6 @@ async def main(client_name: ClientName = "openai_chat") -> None:
     print(f"Client: {client_name}")
     print(f"User: {message.text}")
 
-    if isinstance(client, FoundryChatClient):
-        async with client:
-            if stream:
-                response_stream = client.get_response([message], stream=True, options={"tools": get_weather})
-                print("Assistant: ", end="")
-                async for chunk in response_stream:
-                    if chunk.text:
-                        print(chunk.text, end="")
-                print("")
-            else:
-                print(
-                    f"Assistant: {await client.get_response([message], stream=False, options={'tools': get_weather})}"
-                )
-        return
-
     if stream:
         response_stream = client.get_response([message], stream=True, options={"tools": get_weather})
         print("Assistant: ", end="")
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/README.md b/python/samples/04-hosting/foundry-hosted-agents/responses/README.md
index 072dbea36f..3181cb5ea4 100644
--- a/python/samples/04-hosting/foundry-hosted-agents/responses/README.md
+++ b/python/samples/04-hosting/foundry-hosted-agents/responses/README.md
@@ -8,4 +8,4 @@ This folder contains a list of samples that show how to host agents using the `r
 | [02_local_tools](./02_local_tools) | An example of hosting an agent with the `responses` API and local tools including a function tool and a local shell tool. |
 | [03_remote_mcp](./03_remote_mcp) | An example of hosting an agent with the `responses` API and remote MCPs, including a GitHub MCP server and a Foundry Toolbox. |
 | [04_workflows](./04_workflows) | An example of hosting a workflow with the `responses` API. |
-| [using_deployed_agent.py](./using_deployed_agent.py) | An example of how to use the deployed agent in Agent Framework. |
+| [using_deployed_agent.py](./using_deployed_agent.py) | Connect to the deployed basic Foundry agent with `FoundryAgent`, `allow_preview=True`, and version `v2`. |
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/using_deployed_agent.py b/python/samples/04-hosting/foundry-hosted-agents/responses/using_deployed_agent.py
index 1f3525775a..9d1d50b959 100644
--- a/python/samples/04-hosting/foundry-hosted-agents/responses/using_deployed_agent.py
+++ b/python/samples/04-hosting/foundry-hosted-agents/responses/using_deployed_agent.py
@@ -1,50 +1,146 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
 import asyncio
+import os
+from collections.abc import Mapping
+from typing import Any, cast
+
+from agent_framework import AgentSession
+from agent_framework.foundry import FoundryAgent
+from azure.ai.projects.aio import AIProjectClient
+from azure.ai.projects.models import VersionRefIndicator
+from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
 
-from agent_framework import Agent, AgentResponse, AgentResponseUpdate, ResponseStream
-from agent_framework.openai import OpenAIChatClient
-from typing_extensions import Any
+load_dotenv()
 
 """
-This script demonstrates how to talk to a deployed agent using the OpenAIChatClient.
+This sample demonstrates how to connect to the deployed basic Foundry agent with
+`FoundryAgent`.
+
+The sample uses environment variables for configuration, which can be set in a .env file or in the environment directly:
+Environment variables:
+    FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint.
+    FOUNDRY_AGENT_NAME: Hosted agent name.
+    FOUNDRY_AGENT_VERSION: Hosted agent version. Optional, defaults to latest if not specified.
+
+After you deploy one of the agents in this directory, you can run this sample
+to connect to it and have a conversation.
+
+Note: The `allow_preview=True` flag is required to connect to the new hosted
+agents, as this is a preview feature in Foundry.
 
-Depending on where you have deployed your agent (local or Foundry Hosting), you may
-need to change the base_url when initializing the OpenAIChatClient.
 """
 
 
-async def print_streaming_response(streaming_response: ResponseStream[AgentResponseUpdate, AgentResponse[Any]]) -> None:
-    async for chunk in streaming_response:
-        if chunk.text:
-            print(chunk.text, end="", flush=True)
+async def create_hosted_agent_session(
+    *,
+    agent: FoundryAgent,
+    project_client: AIProjectClient,
+    agent_name: str,
+    agent_version: str | None,
+    isolation_key: str,
+) -> AgentSession:
+    """Create a hosted-agent service session and wrap it in an AgentSession."""
+    create_session_kwargs: dict[str, Any] = {
+        "agent_name": agent_name,
+        "isolation_key": isolation_key,
+    }
+    resolved_agent_version = agent_version
+    if resolved_agent_version is None:
+        agent_details = await cast(Any, project_client.beta.agents).get(  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
+            agent_name=agent_name
+        )
+        versions = getattr(agent_details, "versions", None)
+        if not isinstance(versions, Mapping):
+            raise ValueError("Hosted agent details did not include a versions mapping.")
+        latest_version = getattr(cast(Any, versions.get("latest")), "version", None)
+        if not isinstance(latest_version, str) or not latest_version:
+            raise ValueError("Hosted agent details did not include a latest version string.")
+        resolved_agent_version = latest_version
+
+    create_session_kwargs["version_indicator"] = VersionRefIndicator(agent_version=resolved_agent_version)
+    service_session = await project_client.beta.agents.create_session(**create_session_kwargs)
+    agent_session_id = getattr(service_session, "agent_session_id", None)
+    if not isinstance(agent_session_id, str) or not agent_session_id:
+        raise ValueError("Hosted agent session creation did not return a non-empty agent_session_id.")
+
+    return agent.get_session(agent_session_id)
 
 
 async def main() -> None:
-    agent = Agent(client=OpenAIChatClient(base_url="http://localhost:8088"))
-    session = agent.create_session()
-
-    # First turn
-    query = "Hi!"
-    print(f"User: {query}")
-    print("Agent: ", end="", flush=True)
-    streaming_response = agent.run(query, session=session, stream=True)
-    await print_streaming_response(streaming_response)
-
-    # Second turn
-    query = "Your name is Javis. What can you do?"
-    print(f"\nUser: {query}")
-    print("Agent: ", end="", flush=True)
-    streaming_response = agent.run(query, session=session, stream=True)
-    await print_streaming_response(streaming_response)
-
-    # Third turn
-    query = "What is your name?"
-    print(f"\nUser: {query}")
-    print("Agent: ", end="", flush=True)
-    streaming_response = agent.run(query, session=session, stream=True)
-    await print_streaming_response(streaming_response)
+    credential = AzureCliCredential()
+    project_endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
+    agent_name = os.environ["FOUNDRY_AGENT_NAME"]
+    agent_version = os.getenv("FOUNDRY_AGENT_VERSION")
+    isolation_key = "my-isolation-key"
+
+    project_client = AIProjectClient(
+        endpoint=project_endpoint,
+        credential=credential,
+        allow_preview=True,
+    )
+    async with (
+        project_client,
+        FoundryAgent(
+            project_client=project_client,
+            agent_name=agent_name,
+            agent_version=agent_version,
+            allow_preview=True,
+        ) as agent,
+    ):
+        session = await create_hosted_agent_session(
+            agent=agent,
+            project_client=project_client,
+            agent_name=agent_name,
+            agent_version=agent_version,
+            isolation_key=isolation_key,
+        )
+
+        try:
+            # 1. Send the first turn.
+            query = "Hi!"
+            print(f"User: {query}")
+            print("Agent: ", end="", flush=True)
+            async for chunk in agent.run(query, session=session, stream=True):
+                if chunk.text:
+                    print(chunk.text, end="", flush=True)
+
+            # 2. Continue the conversation with the same deployed agent session.
+            query = "Your name is Javis. What can you do?"
+            print(f"\nUser: {query}")
+            print("Agent: ", end="", flush=True)
+            async for chunk in agent.run(query, session=session, stream=True):
+                if chunk.text:
+                    print(chunk.text, end="", flush=True)
+
+            # 3. Ask a follow-up question in the same session.
+            query = "What is your name?"
+            print(f"\nUser: {query}")
+            print("Agent: ", end="", flush=True)
+            async for chunk in agent.run(query, session=session, stream=True):
+                if chunk.text:
+                    print(chunk.text, end="", flush=True)
+        finally:
+            if session.service_session_id is not None:
+                await project_client.beta.agents.delete_session(
+                    agent_name=agent_name,
+                    session_id=session.service_session_id,
+                    isolation_key=isolation_key,
+                )
 
 
 if __name__ == "__main__":
     asyncio.run(main())
+
+"""
+Sample output:
+User: Hi!
+Agent: Hello! How can I help you today?
+User: Your name is Javis. What can you do?
+Agent: I can answer questions and help with tasks using the instructions configured on the deployed agent.
+User: What is your name?
+Agent: My name is Javis.
+"""