[BREAKING] Python: Ensure session isolation for FHA invocation impl (#7158)

* Ensure session isolation for FHA invocation impl

* Fix type check errors

* Add user isolation to sample

python/CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **agent-framework-azurefunctions**, **agent-framework-core**, **agent-framework-durabletask**: Add HITL response-URL addressing for requests raised from inside workflows ([#7001](https://github.com/microsoft/agent-framework/pull/7001))
 - **agent-framework-core**: Add cross-session origin attribution to context-injected messages ([#7041](https://github.com/microsoft/agent-framework/pull/7041))
 - **agent-framework-core**, **agent-framework-tools**: Warn when auto-approved tools have name collisions ([#7090](https://github.com/microsoft/agent-framework/pull/7090))
+- **agent-framework-core**: Add a `session_provider` option to `MCPSkillsSource` and `MCPSkill` (mutually exclusive with `client`) that resolves the MCP session on every fetch, keeping cached skills reconnect-safe when the underlying session is replaced ([#7135](https://github.com/microsoft/agent-framework/pull/7135))
 - **agent-framework-hosting-a2a**: Add app-owned A2A hosting helpers ([#7050](https://github.com/microsoft/agent-framework/pull/7050))
 - **agent-framework-hosting-mcp**: Add app-owned MCP hosting helpers for exposing agents and workflows as native MCP tools ([#7209](https://github.com/microsoft/agent-framework/pull/7209))
 - **agent-framework-hosting-responses**: [BREAKING] Add Responses conversation ID creation and parsing helpers, and distinguish conversation IDs from previous response IDs ([#7234](https://github.com/microsoft/agent-framework/pull/7234))
@@ -64,6 +65,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **agent-framework-core**: Forward `header_provider` headers to streamable HTTP MCP transports ([#7218](https://github.com/microsoft/agent-framework/pull/7218))
 - **agent-framework-core**: Prevent compaction from emitting empty projections ([#7219](https://github.com/microsoft/agent-framework/pull/7219))
 - **agent-framework-core**: Return MCP tool-use sampling results to the requesting server ([#7189](https://github.com/microsoft/agent-framework/pull/7189))
+- **agent-framework-foundry-hosting**: Make `FoundryToolbox.as_skills_provider()` cache toolbox skill discovery by default so `skill://index.json` is read once instead of on every agent run, give `disable_caching` an observable effect, and add a `cache_refresh_interval` option ([#7135](https://github.com/microsoft/agent-framework/pull/7135))
 - **agent-framework-hosting**, **agent-framework-hosting-responses**: Isolate stored session snapshots from later mutations ([#7141](https://github.com/microsoft/agent-framework/pull/7141))
 - **agent-framework-ollama**: Generate distinct call ids for parallel tool calls ([#6822](https://github.com/microsoft/agent-framework/pull/6822))
 - **agent-framework-orchestrations**: Prevent the Magentic manager from duplicating conversation history ([#6297](https://github.com/microsoft/agent-framework/pull/6297))

python/packages/core/AGENTS.md

@@ -91,7 +91,7 @@ agent_framework/
 - **`SkillScriptRunner`** - Protocol for file-based script execution. Any callable matching `(skill, script, args) -> Any` satisfies it. Code-defined scripts do not use a runner.
 - **`SkillScriptArgumentParser`** - Public type alias for an optional callable `(raw args: dict | list[str] | str | None) -> dict | None` that converts the raw `args` value before an `InlineSkillScript` runs (applied before the inline list-args guard). It is an opt-in customization hook (port of .NET PR #6498) that lets callers support backends sending tool-call arguments in a non-conforming shape (e.g. vLLM JSON strings). The output is constrained to a `dict` (named keyword arguments) or `None`, because inline scripts bind arguments by keyword name. Supply it via the `argument_parser=` constructor arg on `InlineSkillScript`, `InlineSkill` (default for scripts added via `@skill.script`), or `ClassSkill` (default for scripts discovered via `@ClassSkill.script`). When `None` (the default), the raw value is used unchanged. File-based scripts are unaffected (their runner owns arg handling).
 - **`SkillsProvider`** - Context provider (extends `ContextProvider`) that discovers file-based skills from `SKILL.md` files and/or accepts code-defined `Skill` instances. Follows progressive disclosure: advertise → load → read resources / run scripts. By default all three tools it exposes (`load_skill`, `read_skill_resource`, `run_skill_script`) are registered with `approval_mode="always_require"`, so every skill operation needs approval. To run unattended, pass one of the static auto-approval rules to `ToolApprovalMiddleware` (via `auto_approval_rules`): `SkillsProvider.read_only_tools_auto_approval_rule` approves only the read-only tools (`load_skill`, `read_skill_resource`) while still prompting for `run_skill_script`, and `SkillsProvider.all_tools_auto_approval_rule` approves every skill tool including script execution. Both rules reject any call carrying a `server_label` so they stay scoped to this provider's local tools and never auto-approve a same-named hosted tool. Alternatively, for trusted skills, the constructor / `from_paths` kwargs `disable_load_skill_approval`, `disable_read_skill_resource_approval`, and `disable_run_skill_script_approval` (all default `False`) opt individual tools out of approval entirely by registering them with `approval_mode="never_require"` (the auto-approval rules only apply to tools that still require approval). The tool names are also exposed as class constants (`LOAD_SKILL_TOOL_NAME`, `READ_SKILL_RESOURCE_TOOL_NAME`, `RUN_SKILL_SCRIPT_TOOL_NAME`).
-- **`SkillsSource` decorators** - Skill sources are composable: `SkillsSource` is the abstract base, with concrete sources (`InMemorySkillsSource`, `FileSkillsSource`, `MCPSkillsSource`) and decorators that wrap an inner source — `AggregatingSkillsSource` (concatenate several sources), `FilteringSkillsSource` (predicate filter), `DeduplicatingSkillsSource` (first-wins by name), and `CachingSkillsSource` (cache the inner source's skills list). `DelegatingSkillsSource` is the abstract base for decorators. **`get_skills` takes a `SkillsSourceContext`**: every source/decorator implements `async def get_skills(self, context: SkillsSourceContext) -> list[Skill]` and forwards `context` to inner sources. `SkillsSourceContext` (frozen) carries the invoking `agent` (`SupportsAgentRun`) and optional `session` (`AgentSession | None`); `SkillsProvider` builds it from `before_run`'s `agent`/`session` and passes it into the pipeline. `FilteringSkillsSource`'s predicate is context-aware: `Callable[[Skill, SkillsSourceContext], bool]` (port of .NET #6797). **Default caching is applied only to the built-in, context-independent leaf sources**: for the `Skill` / sequence-of-skills / `from_paths` constructors, `SkillsProvider` builds `DeduplicatingSkillsSource(CachingSkillsSource(<file|in-memory leaf>))` so expensive filesystem/network discovery runs once. A **caller-supplied `SkillsSource` is used as-is — never auto-wrapped in caching or deduplication** — because auto-caching a context-aware caller source in a single shared bucket would replay the first invocation's skills for later `SkillsSourceContext`s and leak skills across agents/tenants (matches .NET, whose custom-source constructor also adds no caching/dedup). Callers who want caching on a custom pipeline compose `CachingSkillsSource(inner, cache_isolation_key_selector=...)` themselves. `disable_caching=True` only affects the built-in leaf caching (it has no effect on a caller-supplied source, which is never cached). `CachingSkillsSource` shares a single in-flight fetch across concurrent callers (per cache key) and does not update its cache on a failed fetch, so the next call retries (an initial failure leaves the cache empty; a refresh failure keeps the previously cached list). By default all callers share one cache bucket; pass `cache_isolation_key_selector=Callable[[SkillsSourceContext], str | None]` to cache separately per key (e.g. per agent name) for context-aware inner sources — the key should be low-cardinality and stable, and returning `None` (or leaving the selector `None`) uses the shared bucket. By default a cached list never expires; pass `refresh_interval=timedelta(...)` (port of .NET `CachingAgentSkillsSourceOptions.RefreshInterval`) to treat a cached list as stale once it is older than the interval so the next call re-queries the inner source (useful when an inner source such as `MCPSkillsSource` changes over the process lifetime; a zero/negative interval makes every result immediately stale, and a failed refresh keeps the prior list and retries). Freshness is measured with a monotonic clock (`time.monotonic()`). `SkillsProvider.__init__` / `from_paths` expose a `cache_refresh_interval` kwarg that is threaded into the built-in `CachingSkillsSource` (it has no effect on a caller-supplied source or when `disable_caching=True`).
+- **`SkillsSource` decorators** - Skill sources are composable: `SkillsSource` is the abstract base, with concrete sources (`InMemorySkillsSource`, `FileSkillsSource`, `MCPSkillsSource`) and decorators that wrap an inner source — `AggregatingSkillsSource` (concatenate several sources), `FilteringSkillsSource` (predicate filter), `DeduplicatingSkillsSource` (first-wins by name), and `CachingSkillsSource` (cache the inner source's skills list). `DelegatingSkillsSource` is the abstract base for decorators. **`get_skills` takes a `SkillsSourceContext`**: every source/decorator implements `async def get_skills(self, context: SkillsSourceContext) -> list[Skill]` and forwards `context` to inner sources. `SkillsSourceContext` (frozen) carries the invoking `agent` (`SupportsAgentRun`) and optional `session` (`AgentSession | None`); `SkillsProvider` builds it from `before_run`'s `agent`/`session` and passes it into the pipeline. `FilteringSkillsSource`'s predicate is context-aware: `Callable[[Skill, SkillsSourceContext], bool]` (port of .NET #6797). **Default caching is applied only to the built-in, context-independent leaf sources**: for the `Skill` / sequence-of-skills / `from_paths` constructors, `SkillsProvider` builds `DeduplicatingSkillsSource(CachingSkillsSource(<file|in-memory leaf>))` so expensive filesystem/network discovery runs once. A **caller-supplied `SkillsSource` is used as-is — never auto-wrapped in caching or deduplication** — because auto-caching a context-aware caller source in a single shared bucket would replay the first invocation's skills for later `SkillsSourceContext`s and leak skills across agents/tenants (matches .NET, whose custom-source constructor also adds no caching/dedup). Callers who want caching on a custom pipeline compose `CachingSkillsSource(inner, cache_isolation_key_selector=...)` themselves. `disable_caching=True` only affects the built-in leaf caching (it has no effect on a caller-supplied source, which is never cached). `CachingSkillsSource` shares a single in-flight fetch across concurrent callers (per cache key) and does not update its cache on a failed fetch, so the next call retries (an initial failure leaves the cache empty; a refresh failure keeps the previously cached list). By default all callers share one cache bucket; pass `cache_isolation_key_selector=Callable[[SkillsSourceContext], str | None]` to cache separately per key (e.g. per agent name) for context-aware inner sources — the key should be low-cardinality and stable, and returning `None` (or leaving the selector `None`) uses the shared bucket. By default a cached list never expires; pass `refresh_interval=timedelta(...)` (port of .NET `CachingAgentSkillsSourceOptions.RefreshInterval`) to treat a cached list as stale once it is older than the interval so the next call re-queries the inner source (useful when an inner source such as `MCPSkillsSource` changes over the process lifetime; a zero/negative interval makes every result immediately stale, and a failed refresh keeps the prior list and retries). Freshness is measured with a monotonic clock (`time.monotonic()`). `SkillsProvider.__init__` / `from_paths` expose a `cache_refresh_interval` kwarg that is threaded into the built-in `CachingSkillsSource` (it has no effect on a caller-supplied source or when `disable_caching=True`). **`MCPSkillsSource` and `MCPSkill` accept exactly one of `client` (a fixed `ClientSession`) or `session_provider` (`Callable[[], ClientSession]`, resolved on every fetch); providing both/neither raises `ValueError`.** Use `session_provider` when the underlying session may be swapped over time — e.g. a reconnecting `MCPTool`/`FoundryToolbox` whose `session` is replaced on reconnect — so cached `MCPSkill`s keep fetching against the live session instead of a closed one (`MCPSkillsSource` forwards its provider to every `MCPSkill` it creates). A fixed `client` is safe only when the session outlives the skills.
 
 ### Model Context Protocol (`_mcp.py`)
 

python/packages/core/agent_framework/_skills.py

@@ -4054,6 +4054,40 @@ def _parse_mcp_skill_index(text: str) -> _McpSkillIndex:
     return _McpSkillIndex(schema=raw.get("$schema"), skills=entries)
 
 
+def _resolve_mcp_session_provider(
+    client: ClientSession | None,
+    session_provider: Callable[[], ClientSession] | None,
+) -> Callable[[], ClientSession]:
+    """Normalize the two MCP session inputs into a single session resolver.
+
+    Callers supply **exactly one** of a fixed ``client`` or a
+    ``session_provider`` callable. A fixed client is wrapped in a provider that
+    always returns it; a provider is used as-is so the session is resolved on
+    every call (reconnect-safe for sources whose underlying session is replaced
+    over time, e.g. a reconnecting :class:`~agent_framework.MCPTool`).
+
+    Args:
+        client: A fixed MCP client session, or ``None``.
+        session_provider: A callable returning the current MCP client session,
+            or ``None``.
+
+    Returns:
+        A callable that returns the MCP client session to use.
+
+    Raises:
+        ValueError: If both or neither of *client* and *session_provider* are
+            provided.
+    """
+    if client is not None and session_provider is not None:
+        raise ValueError("Provide exactly one of 'client' or 'session_provider', not both.")
+    if session_provider is not None:
+        return session_provider
+    if client is None:
+        raise ValueError("Provide exactly one of 'client' or 'session_provider'.")
+    fixed: ClientSession = client
+    return lambda: fixed
+
+
 @experimental(feature_id=ExperimentalFeature.MCP_SKILLS)
 class MCPSkillResource(SkillResource):
     """A :class:`SkillResource` backed by content fetched from an MCP server.
@@ -4116,21 +4150,39 @@ class MCPSkill(Skill):
         self,
         frontmatter: SkillFrontmatter,
         skill_md_uri: str,
-        client: ClientSession,
+        client: ClientSession | None = None,
+        *,
+        session_provider: Callable[[], ClientSession] | None = None,
     ) -> None:
         """Initialize an MCPSkill.
 
+        Provide **exactly one** of *client* or *session_provider*.
+
         Args:
             frontmatter: The parsed frontmatter metadata for this skill.
             skill_md_uri: The full MCP resource URI of the ``SKILL.md`` resource
                 (e.g. ``skill://unit-converter/SKILL.md``). The skill's root URI
                 is derived by stripping the trailing ``SKILL.md`` segment.
-            client: The MCP client session used to fetch resources on demand.
+            client: A fixed MCP client session used to fetch resources on demand.
+                Use this when the session outlives the skill (e.g. a caller-owned
+                long-lived session).
+
+        Keyword Args:
+            session_provider: A callable returning the current MCP client session,
+                resolved on every fetch. Prefer this over *client* when the
+                underlying session may be replaced over the skill's lifetime (for
+                example, a reconnecting :class:`~agent_framework.MCPTool` whose
+                ``session`` is swapped on reconnect), so a cached skill keeps
+                using the live session instead of a closed one.
+
+        Raises:
+            ValueError: If both or neither of *client* and *session_provider* are
+                provided.
         """
         self._frontmatter = frontmatter
         self._skill_md_uri = skill_md_uri
         self._skill_root_uri = self._compute_skill_root_uri(skill_md_uri)
-        self._client = client
+        self._session_provider = _resolve_mcp_session_provider(client, session_provider)
         self._content: str | None = None
 
     @property
@@ -4154,7 +4206,7 @@ class MCPSkill(Skill):
         if self._content is not None:
             return self._content
 
-        result = await self._client.read_resource(_mcp_any_url(self._skill_md_uri))
+        result = await self._session_provider().read_resource(_mcp_any_url(self._skill_md_uri))
         text = _mcp_join_text(result)
         if not text:
             raise ValueError(f"The MCP server returned no text content for SKILL.md resource '{self._skill_md_uri}'.")
@@ -4184,7 +4236,7 @@ class MCPSkill(Skill):
 
         uri = self._skill_root_uri + normalized
         try:
-            result = await self._client.read_resource(_mcp_any_url(uri))
+            result = await self._session_provider().read_resource(_mcp_any_url(uri))
         except Exception as ex:
             if _is_mcp_resource_not_found(ex):
                 logger.debug("MCP resource '%s' not available: %s", uri, ex)
@@ -4276,14 +4328,36 @@ class MCPSkillsSource(SkillsSource):
     _INDEX_URI: Final[str] = "skill://index.json"
     _SKILL_MD_TYPE: Final[str] = "skill-md"
 
-    def __init__(self, client: ClientSession) -> None:
+    def __init__(
+        self,
+        client: ClientSession | None = None,
+        *,
+        session_provider: Callable[[], ClientSession] | None = None,
+    ) -> None:
         """Initialize an MCPSkillsSource.
 
+        Provide **exactly one** of *client* or *session_provider*.
+
         Args:
-            client: An MCP client session connected to a server that
-                exposes Agent Skills resources.
+            client: A fixed MCP client session connected to a server that exposes
+                Agent Skills resources. Use this when the session outlives the
+                source (e.g. a caller-owned long-lived session).
+
+        Keyword Args:
+            session_provider: A callable returning the current MCP client session,
+                resolved on every discovery and on each skill's on-demand fetch.
+                Prefer this over *client* when the underlying session may be
+                replaced over time (for example, a reconnecting
+                :class:`~agent_framework.MCPTool` whose ``session`` is swapped on
+                reconnect), so cached skills keep using the live session. The
+                provider is forwarded to every :class:`MCPSkill` this source
+                creates.
+
+        Raises:
+            ValueError: If both or neither of *client* and *session_provider* are
+                provided.
         """
-        self._client = client
+        self._session_provider = _resolve_mcp_session_provider(client, session_provider)
 
     async def get_skills(self, context: SkillsSourceContext) -> list[Skill]:
         """Discover and return skills from the MCP server.
@@ -4327,7 +4401,7 @@ class MCPSkillsSource(SkillsSource):
             absent, empty, or malformed.
         """
         try:
-            result = await self._client.read_resource(_mcp_any_url(self._INDEX_URI))
+            result = await self._session_provider().read_resource(_mcp_any_url(self._INDEX_URI))
         except Exception as ex:
             if _is_mcp_resource_not_found(ex):
                 logger.debug("No skill://index.json resource available on MCP server: %s", ex)
@@ -4382,7 +4456,7 @@ class MCPSkillsSource(SkillsSource):
             logger.debug("Skipping entry '%s': invalid metadata: %s", entry.name, ex)
             return None
 
-        return MCPSkill(frontmatter=fm, skill_md_uri=entry.url, client=self._client)
+        return MCPSkill(frontmatter=fm, skill_md_uri=entry.url, session_provider=self._session_provider)
 
 
 # endregion

python/packages/core/tests/core/test_mcp_skills.py

@@ -159,14 +159,12 @@ class TestMCPSkillsExperimentalStage:
 class TestMCPSkillResource:
     """Tests for MCPSkillResource."""
 
-    @pytest.mark.asyncio
     async def test_read_text_content(self) -> None:
         result = _make_text_result("hello world")
         resource = MCPSkillResource(name="test.md", result=result)
         content = await resource.read()
         assert content == "hello world"
 
-    @pytest.mark.asyncio
     async def test_read_binary_content(self) -> None:
         data = bytes([0x01, 0x02, 0x03, 0x04])
         result = _make_blob_result(data)
@@ -174,14 +172,12 @@ class TestMCPSkillResource:
         content = await resource.read()
         assert content == data
 
-    @pytest.mark.asyncio
     async def test_read_empty_returns_none(self) -> None:
         result = _make_empty_result()
         resource = MCPSkillResource(name="empty", result=result)
         content = await resource.read()
         assert content is None
 
-    @pytest.mark.asyncio
     async def test_read_multiple_text_contents_joined(self) -> None:
         result = ReadResourceResult(
             contents=[
@@ -193,7 +189,6 @@ class TestMCPSkillResource:
         content = await resource.read()
         assert content == "line1\nline2"
 
-    @pytest.mark.asyncio
     async def test_binary_takes_precedence_over_text(self) -> None:
         data = b"\xff\xfe"
         result = ReadResourceResult(
@@ -221,7 +216,6 @@ class TestMCPSkillResource:
 class TestMCPSkill:
     """Tests for MCPSkill."""
 
-    @pytest.mark.asyncio
     async def test_get_content_fetches_and_caches(self) -> None:
         client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD)})
         from agent_framework import SkillFrontmatter
@@ -237,7 +231,6 @@ class TestMCPSkill:
         # Only one MCP call should be made (cached)
         assert client.read_resource.call_count == 1
 
-    @pytest.mark.asyncio
     async def test_get_content_raises_on_empty(self) -> None:
         client = _make_client(**{"skill://empty/SKILL.md": _make_empty_result()})
         from agent_framework import SkillFrontmatter
@@ -248,7 +241,6 @@ class TestMCPSkill:
         with pytest.raises(ValueError, match="no text content"):
             await skill.get_content()
 
-    @pytest.mark.asyncio
     async def test_get_resource_text(self) -> None:
         client = _make_client(**{
             "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
@@ -264,7 +256,6 @@ class TestMCPSkill:
         content = await resource.read()
         assert content == "- check thing 1\n- check thing 2"
 
-    @pytest.mark.asyncio
     async def test_get_resource_binary(self) -> None:
         data = bytes([0x01, 0x02, 0x03, 0x04])
         client = _make_client(**{
@@ -281,7 +272,6 @@ class TestMCPSkill:
         content = await resource.read()
         assert content == data
 
-    @pytest.mark.asyncio
     async def test_get_resource_unknown_returns_none(self) -> None:
         client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD)})
         from agent_framework import SkillFrontmatter
@@ -292,7 +282,6 @@ class TestMCPSkill:
         resource = await skill.get_resource("references/does-not-exist.md")
         assert resource is None
 
-    @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "name",
         [
@@ -320,7 +309,6 @@ class TestMCPSkill:
         assert resource is None
         client.read_resource.assert_not_called()
 
-    @pytest.mark.asyncio
     async def test_get_resource_empty_name_returns_none(self) -> None:
         client = _make_client()
         from agent_framework import SkillFrontmatter
@@ -331,7 +319,6 @@ class TestMCPSkill:
         assert await skill.get_resource("") is None
         assert await skill.get_resource("   ") is None
 
-    @pytest.mark.asyncio
     async def test_get_script_returns_none(self) -> None:
         client = _make_client()
         from agent_framework import SkillFrontmatter
@@ -350,6 +337,46 @@ class TestMCPSkill:
     def test_compute_skill_root_uri_no_suffix_adds_slash(self) -> None:
         assert MCPSkill._compute_skill_root_uri("skill://unit-converter") == "skill://unit-converter/"
 
+    async def test_session_provider_resolves_live_session(self) -> None:
+        # A session_provider is resolved on every fetch, so a skill built against
+        # one session follows a reconnect that swaps the session object.
+        from agent_framework import SkillFrontmatter
+
+        old_client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result("# Old\nold body")})
+        new_client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result("# New\nnew body")})
+        current = {"session": old_client}
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(
+            frontmatter=fm,
+            skill_md_uri="skill://unit-converter/SKILL.md",
+            session_provider=lambda: current["session"],
+        )
+
+        # Swap the session (as a reconnect would) before the first fetch.
+        current["session"] = new_client
+        content = await skill.get_content()
+
+        assert "new body" in content
+        old_client.read_resource.assert_not_called()
+        new_client.read_resource.assert_called_once()
+
+    def test_requires_exactly_one_of_client_or_session_provider(self) -> None:
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        client = _make_client()
+
+        with pytest.raises(ValueError, match="exactly one"):
+            MCPSkill(frontmatter=fm, skill_md_uri="skill://x/SKILL.md")
+        with pytest.raises(ValueError, match="exactly one"):
+            MCPSkill(
+                frontmatter=fm,
+                skill_md_uri="skill://x/SKILL.md",
+                client=client,
+                session_provider=lambda: client,
+            )
+
 
 # ---------------------------------------------------------------------------
 # MCPSkillsSource tests
@@ -359,7 +386,6 @@ class TestMCPSkill:
 class TestMCPSkillsSource:
     """Tests for MCPSkillsSource."""
 
-    @pytest.mark.asyncio
     async def test_index_based_discovery_returns_skill(self) -> None:
         client = _make_client(**{
             "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
@@ -376,14 +402,12 @@ class TestMCPSkillsSource:
         content = await skills[0].get_content()
         assert "Body content here." in content
 
-    @pytest.mark.asyncio
     async def test_no_index_returns_empty(self) -> None:
         client = _make_client()  # No resources at all
         source = MCPSkillsSource(client=client)
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_does_not_read_skill_md_during_discovery(self) -> None:
         # Index points to a skill, but SKILL.md is not registered on the server.
         # Discovery should succeed because it only reads the index.
@@ -394,7 +418,6 @@ class TestMCPSkillsSource:
         assert len(skills) == 1
         assert skills[0].frontmatter.name == "unit-converter"
 
-    @pytest.mark.asyncio
     async def test_invalid_name_is_skipped(self) -> None:
         index_json = json.dumps({
             "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
@@ -412,7 +435,6 @@ class TestMCPSkillsSource:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_missing_required_fields_is_skipped(self) -> None:
         index_json = json.dumps({
             "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
@@ -429,7 +451,6 @@ class TestMCPSkillsSource:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_unsupported_type_is_skipped(self) -> None:
         index_json = json.dumps({
             "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
@@ -447,7 +468,6 @@ class TestMCPSkillsSource:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_template_type_is_skipped(self) -> None:
         index_json = json.dumps({
             "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
@@ -464,21 +484,18 @@ class TestMCPSkillsSource:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_empty_index_returns_empty(self) -> None:
         client = _make_client(**{"skill://index.json": _make_text_result('{"skills": []}', uri="skill://index.json")})
         source = MCPSkillsSource(client=client)
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_malformed_index_json_returns_empty(self) -> None:
         client = _make_client(**{"skill://index.json": _make_text_result("not valid json", uri="skill://index.json")})
         source = MCPSkillsSource(client=client)
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_sibling_text_resource(self) -> None:
         client = _make_client(**{
             "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
@@ -492,7 +509,6 @@ class TestMCPSkillsSource:
         content = await resource.read()
         assert content == "- check thing 1\n- check thing 2"
 
-    @pytest.mark.asyncio
     async def test_sibling_binary_resource(self) -> None:
         data = bytes([0x01, 0x02, 0x03, 0x04])
         client = _make_client(**{
@@ -507,6 +523,36 @@ class TestMCPSkillsSource:
         content = await resource.read()
         assert content == data
 
+    async def test_session_provider_resolves_live_session(self) -> None:
+        # Discovery and the resulting skills' on-demand fetches both resolve the
+        # provider, so a source built before a reconnect follows the swapped session.
+        old_client = _make_client(**{
+            "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
+            "skill://unit-converter/SKILL.md": _make_text_result("# Old\nold body"),
+        })
+        new_client = _make_client(**{
+            "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
+            "skill://unit-converter/SKILL.md": _make_text_result("# New\nnew body"),
+        })
+        current = {"session": old_client}
+
+        source = MCPSkillsSource(session_provider=lambda: current["session"])
+        skills = await source.get_skills(_SOURCE_CTX)
+        assert len(skills) == 1
+
+        # A reconnect swaps the session; the already-discovered skill must fetch
+        # its content from the new session, not the closed one.
+        current["session"] = new_client
+        content = await skills[0].get_content()
+        assert "new body" in content
+
+    def test_requires_exactly_one_of_client_or_session_provider(self) -> None:
+        client = _make_client()
+        with pytest.raises(ValueError, match="exactly one"):
+            MCPSkillsSource()
+        with pytest.raises(ValueError, match="exactly one"):
+            MCPSkillsSource(client=client, session_provider=lambda: client)
+
 
 # ---------------------------------------------------------------------------
 # McpError code branching tests
@@ -522,7 +568,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
     crashes, and connection drops are visible.
     """
 
-    @pytest.mark.asyncio
     async def test_index_method_not_found_returns_empty(self) -> None:
         """METHOD_NOT_FOUND (-32601) -> server doesn't support resources/read."""
         client = AsyncMock()
@@ -531,7 +576,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_index_resource_not_found_returns_empty(self) -> None:
         """MCP-spec "Resource not found" (-32002) -> server has no index."""
         client = AsyncMock()
@@ -542,7 +586,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         skills = await source.get_skills(_SOURCE_CTX)
         assert skills == []
 
-    @pytest.mark.asyncio
     async def test_index_invalid_params_propagates(self) -> None:
         """INVALID_PARAMS (-32602) is a real bug, must propagate (not "not found")."""
         client = AsyncMock()
@@ -551,7 +594,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await source.get_skills(_SOURCE_CTX)
 
-    @pytest.mark.asyncio
     async def test_index_internal_error_propagates(self) -> None:
         """INTERNAL_ERROR (-32603) must propagate, not silently return empty."""
         client = AsyncMock()
@@ -560,7 +602,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await source.get_skills(_SOURCE_CTX)
 
-    @pytest.mark.asyncio
     async def test_index_connection_closed_propagates(self) -> None:
         """CONNECTION_CLOSED (-32000) must propagate."""
         client = AsyncMock()
@@ -571,7 +612,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await source.get_skills(_SOURCE_CTX)
 
-    @pytest.mark.asyncio
     async def test_index_generic_error_code_propagates(self) -> None:
         """Generic handler error (code 0) must propagate."""
         client = AsyncMock()
@@ -580,7 +620,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await source.get_skills(_SOURCE_CTX)
 
-    @pytest.mark.asyncio
     async def test_index_non_mcp_error_propagates(self) -> None:
         """Non-McpError exceptions (connection drop, timeout) must propagate."""
         client = AsyncMock()
@@ -589,7 +628,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(ConnectionError):
             await source.get_skills(_SOURCE_CTX)
 
-    @pytest.mark.asyncio
     async def test_get_resource_internal_error_propagates(self) -> None:
         """McpError with INTERNAL_ERROR on get_resource must propagate."""
         from agent_framework import SkillFrontmatter
@@ -601,7 +639,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await skill.get_resource("references/file.md")
 
-    @pytest.mark.asyncio
     async def test_get_resource_not_found_returns_none(self) -> None:
         """McpError with RESOURCE_NOT_FOUND (-32002) on get_resource returns None."""
         from agent_framework import SkillFrontmatter
@@ -615,7 +652,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         result = await skill.get_resource("references/file.md")
         assert result is None
 
-    @pytest.mark.asyncio
     async def test_get_resource_connection_error_propagates(self) -> None:
         """A plain ConnectionError on get_resource must propagate, not return None."""
         from agent_framework import SkillFrontmatter
@@ -627,7 +663,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(ConnectionError):
             await skill.get_resource("references/file.md")
 
-    @pytest.mark.asyncio
     async def test_get_resource_timeout_error_propagates(self) -> None:
         """A TimeoutError on get_resource must propagate, not return None."""
         from agent_framework import SkillFrontmatter
@@ -639,7 +674,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(TimeoutError):
             await skill.get_resource("references/file.md")
 
-    @pytest.mark.asyncio
     async def test_get_resource_generic_mcp_error_propagates(self) -> None:
         """McpError with a generic code (0) on get_resource must propagate."""
         from agent_framework import SkillFrontmatter
@@ -651,7 +685,6 @@ class TestMCPSkillsSourceErrorCodeBranching:
         with pytest.raises(McpError):
             await skill.get_resource("references/file.md")
 
-    @pytest.mark.asyncio
     async def test_index_timeout_error_propagates(self) -> None:
         """A TimeoutError reading skill://index.json must propagate."""
         client = AsyncMock()

python/packages/foundry_hosting/agent_framework_foundry_hosting/_invocations.py

@@ -1,9 +1,10 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from agent_framework import AgentSession, BaseAgent, SupportsAgentRun
+from agent_framework import AgentSession, SupportsAgentRun
+from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.invocations import InvocationAgentServerHost
 from starlette.requests import Request
-from starlette.responses import JSONResponse, Response, StreamingResponse
+from starlette.responses import Response, StreamingResponse
 from typing_extensions import Any, AsyncGenerator
 
 
@@ -12,7 +13,7 @@ class InvocationsHostServer(InvocationAgentServerHost):
 
     def __init__(
         self,
-        agent: BaseAgent,
+        agent: SupportsAgentRun,
         *,
         openapi_spec: dict[str, Any] | None = None,
         **kwargs: Any,
@@ -30,17 +31,57 @@ class InvocationsHostServer(InvocationAgentServerHost):
         """
         super().__init__(openapi_spec=openapi_spec, **kwargs)
 
-        if not isinstance(agent, SupportsAgentRun):
-            raise TypeError("Agent must support the SupportsAgentRun interface")
-
         self._agent = agent
         self._sessions: dict[str, AgentSession] = {}
         self.invoke_handler(self._handle_invoke)
 
+    def _partition_key(self) -> str:
+        """Get the partition key for the current request.
+
+        A partition key is made up of the session ID and user ID. If the request is not
+        from a hosted environment, the partition key will be just the session ID. In the
+        Foundry hosted environment, the partition key is used to maintain isolation between
+        different sessions and users, such that one user cannot access another user's sessions.
+
+        Returns:
+            The partition key for the current request.
+
+        Exceptions:
+            RuntimeError: If the context doesn't contain the expected IDs.
+        """
+        context = get_request_context()
+
+        # Fail fast if the service is on protocol v1.0.0
+        if self.config.is_hosted and context.call_id is None:
+            raise RuntimeError(
+                "The hosted environment is running on protocol 1.0.0, but the agent requires protocol 2.0.0. "
+                "Please upgrade your agent protocol to 2.0.0 in `agent.manifest.yaml` or `agent.yaml`, or "
+                "downgrade the `agent-framework-foundry-hosting` package to `1.0.0a260625` or before to use 1.0.0."
+            )
+
+        if self.config.is_hosted:
+            if not context.session_id or not context.user_id:
+                raise RuntimeError(
+                    "The hosted environment is missing session_id or user_id in the request context. "
+                    "Please ensure that the request is coming from a valid Foundry platform service."
+                )
+            return f"{context.session_id}:{context.user_id}"
+
+        if not context.session_id:
+            raise RuntimeError(
+                "The request context is missing session_id. Please ensure that the request is a valid request."
+            )
+
+        return context.session_id
+
     async def _handle_invoke(self, request: Request) -> Response:
         """Invoke the agent with the given request."""
+        try:
+            session_id = self._partition_key()
+        except Exception as e:
+            return Response(content=str(e), status_code=500)
+
         data = await request.json()
-        session_id: str = request.state.session_id
 
         stream = data.get("stream", False)
         user_message = data.get("message", None)
@@ -65,8 +106,5 @@ class InvocationsHostServer(InvocationAgentServerHost):
                 headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
             )
 
-        response = await self._agent.run([user_message], session=session, stream=stream)
-        return JSONResponse({
-            "response": response.text,
-            "session_id": session_id,
-        })
+        response = await self._agent.run([user_message], session=session)
+        return Response(content=response.text)

python/packages/foundry_hosting/agent_framework_foundry_hosting/_toolbox.py

@@ -9,6 +9,8 @@ from urllib.parse import urlsplit
 
 import httpx
 from agent_framework import (
+    CachingSkillsSource,
+    DeduplicatingSkillsSource,
     MCPSkillsSource,
     MCPStreamableHTTPTool,
     SkillsProvider,
@@ -19,9 +21,11 @@ from azure.ai.agentserver.core import get_request_context
 
 if TYPE_CHECKING:
     from collections.abc import Generator
+    from datetime import timedelta
 
     from agent_framework import Skill
     from azure.core.credentials import TokenCredential
+    from mcp.client.session import ClientSession
 
 logger = logging.getLogger(__name__)
 
@@ -195,6 +199,7 @@ class FoundryToolbox(MCPStreamableHTTPTool):
         source_id: str | None = None,
         instruction_template: str | None = None,
         disable_caching: bool = False,
+        cache_refresh_interval: timedelta | None = None,
         disable_load_skill_approval: bool = False,
         disable_read_skill_resource_approval: bool = False,
         disable_run_skill_script_approval: bool = False,
@@ -215,8 +220,18 @@ class FoundryToolbox(MCPStreamableHTTPTool):
             source_id: Unique identifier for the provider instance.
             instruction_template: Custom system-prompt template for advertising
                 skills; see :class:`~agent_framework.SkillsProvider`.
-            disable_caching: Re-query the toolbox on every agent run instead of
-                caching after the first discovery.
+            disable_caching: When ``True``, re-query the toolbox on every agent run,
+                re-reading ``skill://index.json`` each time. When ``False`` (the
+                default), the toolbox's skill discovery is cached after the first run
+                so the index is read once. The toolbox's advertised skill set is the
+                same for every caller (the per-request call-id governs execution/
+                authorization, not which skills are listed), so a single shared cache
+                is safe.
+            cache_refresh_interval: Optional duration after which the cached skill
+                discovery is considered stale and re-read from the toolbox on the next
+                agent run. Useful when a toolbox's attached skills change over the
+                process lifetime. When ``None`` (the default), the cache never expires.
+                Ignored when ``disable_caching=True``.
             disable_load_skill_approval: When ``True``, register the provider's
                 ``load_skill`` tool with ``approval_mode="never_require"`` so loading
                 a skill body needs no host approval. Set this for unattended agents
@@ -250,11 +265,17 @@ class FoundryToolbox(MCPStreamableHTTPTool):
                 )
                 await ResponsesHostServer(agent).run_async()
         """
+        # The toolbox advertises the same skill set to every caller (the per-request
+        # call-id governs execution/authorization, not which skills are listed), so a
+        # single shared cache is safe. SkillsProvider won't auto-cache a caller source,
+        # so we compose the caching ourselves.
+        source: SkillsSource = _FoundryToolboxSkillsSource(self)
+        if not disable_caching:
+            source = DeduplicatingSkillsSource(CachingSkillsSource(source, refresh_interval=cache_refresh_interval))
         return SkillsProvider(
-            _FoundryToolboxSkillsSource(self),
+            source,
             source_id=source_id,
             instruction_template=instruction_template,
-            disable_caching=disable_caching,
             disable_load_skill_approval=disable_load_skill_approval,
             disable_read_skill_resource_approval=disable_read_skill_resource_approval,
             disable_run_skill_script_approval=disable_run_skill_script_approval,
@@ -265,14 +286,17 @@ class _FoundryToolboxSkillsSource(SkillsSource):
     """Discovers skills from a connected :class:`FoundryToolbox` MCP session.
 
     The toolbox's MCP ``session`` is established lazily when the toolbox connects
-    (via the agent or an ``async with`` block), so the session is resolved at
-    discovery time rather than captured at construction.
+    (via the agent or an ``async with`` block) and is **replaced** with a new
+    object whenever the toolbox reconnects. Skills are therefore bound to a
+    ``session_provider`` that resolves the toolbox's current session on every
+    fetch, so cached skills keep using the live session instead of a closed one.
     """
 
     def __init__(self, toolbox: FoundryToolbox) -> None:
         self._toolbox = toolbox
 
-    async def get_skills(self, context: SkillsSourceContext) -> list[Skill]:
+    def _require_session(self) -> ClientSession:
+        """Return the toolbox's current MCP session, or raise if not connected."""
         session = self._toolbox.session
         if session is None:
             raise RuntimeError(
@@ -280,4 +304,10 @@ class _FoundryToolboxSkillsSource(SkillsSource):
                 "Pass the toolbox to the agent (tools=...) or enter it as an async "
                 "context manager before the agent runs."
             )
-        return await MCPSkillsSource(client=session).get_skills(context)
+        return session
+
+    async def get_skills(self, context: SkillsSourceContext) -> list[Skill]:
+        # Fail fast at discovery if not connected, then hand the source a provider
+        # (not a fixed session) so skills survive a reconnect that swaps the session.
+        self._require_session()
+        return await MCPSkillsSource(session_provider=self._require_session).get_skills(context)

python/packages/foundry_hosting/tests/test_invocations.py

@@ -0,0 +1,272 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Unit tests for InvocationsHostServer.
+
+These tests exercise ``InvocationsHostServer`` directly by constructing the
+host, driving ``_partition_key`` and ``_handle_invoke`` with a fake agent and
+mock requests. The Foundry request context is injected via the public
+``set_request_context`` / ``reset_request_context`` helpers rather than by
+patching, matching the style used in ``test_toolbox.py``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from agent_framework import (
+    AgentResponse,
+    AgentResponseUpdate,
+    AgentSession,
+    Content,
+    Message,
+    ServiceSessionId,
+)
+from azure.ai.agentserver.core import (
+    FoundryAgentRequestContext,
+    reset_request_context,
+    set_request_context,
+)
+from starlette.requests import Request
+from starlette.responses import Response, StreamingResponse
+from typing_extensions import Any
+
+from agent_framework_foundry_hosting import InvocationsHostServer
+
+# region Helpers
+
+
+class _FakeAgent:
+    """Minimal agent implementing the ``SupportsAgentRun`` protocol.
+
+    ``run`` returns an awaitable when ``stream`` is ``False`` and an async
+    iterator when ``stream`` is ``True``. Call arguments are recorded on
+    ``calls`` for assertions.
+    """
+
+    def __init__(
+        self,
+        *,
+        response: AgentResponse | None = None,
+        stream_updates: list[AgentResponseUpdate] | None = None,
+    ) -> None:
+        self.id = "fake-agent"
+        self.name: str | None = "Fake Agent"
+        self.description: str | None = "A fake agent for testing"
+        self._response = response
+        self._stream_updates = stream_updates or []
+        self.calls: list[dict[str, Any]] = []
+
+    def run(
+        self,
+        messages: Any = None,
+        *,
+        stream: bool = False,
+        session: AgentSession | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self.calls.append({"messages": messages, "stream": stream, "session": session})
+        if stream:
+
+            async def _gen() -> AsyncIterator[AgentResponseUpdate]:
+                for update in self._stream_updates:
+                    yield update
+
+            return _gen()
+
+        async def _run() -> AgentResponse:
+            assert self._response is not None
+            return self._response
+
+        return _run()
+
+    def create_session(self, *, session_id: str | None = None) -> AgentSession:
+        return AgentSession(session_id=session_id)
+
+    def get_session(
+        self,
+        service_session_id: str | ServiceSessionId,
+        *,
+        session_id: str | None = None,
+    ) -> AgentSession:
+        return AgentSession(service_session_id=service_session_id, session_id=session_id)
+
+
+def _make_agent(
+    *,
+    response_text: str | None = None,
+    stream_texts: list[str] | None = None,
+) -> _FakeAgent:
+    """Build a ``_FakeAgent`` from plain text for non-streaming/streaming runs."""
+    response = None
+    if response_text is not None:
+        response = AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text(response_text)])])
+    stream_updates = None
+    if stream_texts is not None:
+        stream_updates = [AgentResponseUpdate(contents=[Content.from_text(t)]) for t in stream_texts]
+    return _FakeAgent(response=response, stream_updates=stream_updates)
+
+
+def _make_request(payload: dict[str, Any]) -> Request:
+    """Build a mock Starlette request whose ``json()`` returns ``payload``."""
+    request = MagicMock(spec=Request)
+    request.json = AsyncMock(return_value=payload)
+    return request
+
+
+@contextmanager
+def _request_context(
+    *,
+    call_id: str | None = None,
+    user_id: str | None = None,
+    session_id: str | None = None,
+) -> Iterator[None]:
+    """Install a Foundry request context for the duration of the block."""
+    token = set_request_context(FoundryAgentRequestContext(call_id=call_id, user_id=user_id, session_id=session_id))
+    try:
+        yield
+    finally:
+        reset_request_context(token)
+
+
+async def _collect_stream(response: StreamingResponse) -> str:
+    """Concatenate the string chunks produced by a StreamingResponse."""
+    chunks: list[str] = []
+    async for chunk in response.body_iterator:
+        chunks.append(chunk if isinstance(chunk, str) else bytes(chunk).decode())
+    return "".join(chunks)
+
+
+# endregion
+
+
+# region Initialization
+
+
+class TestInit:
+    def test_accepts_supports_agent_run(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        assert server._agent is not None  # pyright: ignore[reportPrivateUsage]
+        assert server._sessions == {}  # pyright: ignore[reportPrivateUsage]
+
+
+# endregion
+
+
+# region Partition key
+
+
+class TestPartitionKey:
+    def test_local_returns_session_id(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        with _request_context(session_id="sess-1"):
+            assert server._partition_key() == "sess-1"  # pyright: ignore[reportPrivateUsage]
+
+    def test_local_missing_session_id_raises(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        with _request_context(), pytest.raises(RuntimeError, match="missing session_id"):
+            server._partition_key()  # pyright: ignore[reportPrivateUsage]
+
+    def test_hosted_without_call_id_raises_protocol_error(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        server.config.is_hosted = True
+        with (
+            _request_context(session_id="sess-1", user_id="user-1"),
+            pytest.raises(RuntimeError, match="protocol 2.0.0"),
+        ):
+            server._partition_key()  # pyright: ignore[reportPrivateUsage]
+
+    def test_hosted_missing_user_id_raises(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        server.config.is_hosted = True
+        with (
+            _request_context(call_id="call-1", session_id="sess-1"),
+            pytest.raises(RuntimeError, match="missing session_id or user_id"),
+        ):
+            server._partition_key()  # pyright: ignore[reportPrivateUsage]
+
+    def test_hosted_returns_composite_key(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        server.config.is_hosted = True
+        with _request_context(call_id="call-1", session_id="sess-1", user_id="user-1"):
+            assert server._partition_key() == "sess-1:user-1"  # pyright: ignore[reportPrivateUsage]
+
+
+# endregion
+
+
+# region Handle invoke
+
+
+class TestHandleInvoke:
+    async def test_missing_message_returns_400(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        request = _make_request({"stream": False})
+        with _request_context(session_id="sess-1"):
+            response = await server._handle_invoke(request)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(response, Response)
+        assert response.status_code == 400
+
+    async def test_missing_message_streaming_returns_400(self) -> None:
+        server = InvocationsHostServer(_make_agent(stream_texts=["a"]))
+        request = _make_request({"stream": True})
+        with _request_context(session_id="sess-1"):
+            response = await server._handle_invoke(request)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(response, StreamingResponse)
+        assert response.status_code == 400
+
+    async def test_partition_key_failure_returns_500(self) -> None:
+        server = InvocationsHostServer(_make_agent(response_text="hi"))
+        request = _make_request({"message": "Hi"})
+        # No session_id in the (local) context -> _partition_key raises -> 500.
+        with _request_context():
+            response = await server._handle_invoke(request)  # pyright: ignore[reportPrivateUsage]
+        assert isinstance(response, Response)
+        assert response.status_code == 500
+
+    async def test_non_streaming_returns_agent_text(self) -> None:
+        agent = _make_agent(response_text="Hello!")
+        server = InvocationsHostServer(agent)
+        request = _make_request({"message": "Hi", "stream": False})
+        with _request_context(session_id="sess-1"):
+            response = await server._handle_invoke(request)  # pyright: ignore[reportPrivateUsage]
+
+        assert isinstance(response, Response)
+        assert response.status_code == 200
+        assert bytes(response.body).decode() == "Hello!"
+        # Agent is called with the message wrapped in a list and the cached session.
+        assert agent.calls[0]["messages"] == ["Hi"]
+        assert agent.calls[0]["stream"] is False
+        assert agent.calls[0]["session"] is server._sessions["sess-1"]  # pyright: ignore[reportPrivateUsage]
+
+    async def test_streaming_yields_update_text(self) -> None:
+        agent = _make_agent(stream_texts=["Hel", "lo", "!"])
+        server = InvocationsHostServer(agent)
+        request = _make_request({"message": "Hi", "stream": True})
+        with _request_context(session_id="sess-1"):
+            response = await server._handle_invoke(request)  # pyright: ignore[reportPrivateUsage]
+
+        assert isinstance(response, StreamingResponse)
+        assert response.media_type == "text/event-stream"
+        assert await _collect_stream(response) == "Hello!"
+        assert agent.calls[0]["messages"] == "Hi"
+        assert agent.calls[0]["stream"] is True
+
+    async def test_session_is_reused_across_requests(self) -> None:
+        agent = _make_agent(response_text="ok")
+        server = InvocationsHostServer(agent)
+
+        with _request_context(session_id="sess-1"):
+            await server._handle_invoke(_make_request({"message": "one"}))  # pyright: ignore[reportPrivateUsage]
+            first_session = server._sessions["sess-1"]  # pyright: ignore[reportPrivateUsage]
+            await server._handle_invoke(_make_request({"message": "two"}))  # pyright: ignore[reportPrivateUsage]
+            second_session = server._sessions["sess-1"]  # pyright: ignore[reportPrivateUsage]
+
+        assert first_session is second_session
+        assert list(server._sessions) == ["sess-1"]  # pyright: ignore[reportPrivateUsage]
+        assert agent.calls[0]["session"] is agent.calls[1]["session"]
+
+
+# endregion

python/packages/foundry_hosting/tests/test_invocations_int.py

@@ -0,0 +1,242 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Integration tests for InvocationsHostServer with a real Foundry endpoint.
+
+These tests exercise the full HTTP pipeline using httpx.AsyncClient with
+ASGITransport — no real server process is started. The agent talks to a real
+Foundry project endpoint so every test requires valid credentials.
+
+The invocations protocol is intentionally simple: a request is a JSON body with
+a ``message`` field (and an optional ``stream`` flag). Non-streaming responses
+return the agent's answer as plain text; streaming responses return the answer
+as a ``text/event-stream`` of text chunks. Session continuity is keyed off the
+``agent_session_id`` query parameter.
+
+Required environment variables:
+    FOUNDRY_PROJECT_ENDPOINT - The Azure AI Foundry project endpoint URL.
+    FOUNDRY_MODEL            - The model deployment name (e.g. gpt-4o).
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Annotated, Any
+
+import httpx
+import pytest
+from agent_framework import Agent, tool
+from agent_framework.foundry import FoundryChatClient
+from azure.identity import AzureCliCredential
+
+from agent_framework_foundry_hosting import InvocationsHostServer
+
+# ---------------------------------------------------------------------------
+# Skip / marker helpers
+# ---------------------------------------------------------------------------
+
+skip_if_foundry_hosting_integration_tests_disabled = pytest.mark.skipif(
+    os.getenv("FOUNDRY_PROJECT_ENDPOINT", "") in ("", "https://test-project.services.ai.azure.com/")
+    or os.getenv("FOUNDRY_MODEL", "") == "",
+    reason="No real FOUNDRY_PROJECT_ENDPOINT or FOUNDRY_MODEL provided; skipping integration tests.",
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def server() -> InvocationsHostServer:
+    """Create an InvocationsHostServer backed by a real Foundry agent."""
+    client = FoundryChatClient(credential=AzureCliCredential())  # pyrefly: ignore[bad-argument-type]
+
+    agent = Agent(
+        client=client,  # ty: ignore[invalid-argument-type]
+        instructions="You are a concise assistant. Keep answers very short (one or two sentences).",
+        default_options={"store": False},  # pyrefly: ignore[bad-argument-type]
+    )
+
+    return InvocationsHostServer(agent)
+
+
+@tool
+async def get_weather(location: Annotated[str, "The city name"]) -> str:
+    """Get the current weather in a given location."""
+    return f"The weather in {location} is 72°F and sunny."
+
+
+@pytest.fixture
+def server_with_tools() -> InvocationsHostServer:
+    """Create an InvocationsHostServer whose agent has a tool."""
+    client = FoundryChatClient(credential=AzureCliCredential())  # pyrefly: ignore[bad-argument-type]
+
+    agent = Agent(
+        client=client,  # ty: ignore[invalid-argument-type]
+        instructions="You are a concise assistant. Use the provided tools when appropriate. Keep answers very short.",
+        tools=[get_weather],
+        default_options={"store": False},  # pyrefly: ignore[bad-argument-type]
+    )
+
+    return InvocationsHostServer(agent)
+
+
+# ---------------------------------------------------------------------------
+# HTTP helpers
+# ---------------------------------------------------------------------------
+
+
+async def _post_invocation(
+    server: InvocationsHostServer,
+    *,
+    message: str,
+    stream: bool = False,
+    session_id: str | None = None,
+) -> httpx.Response:
+    """Send a POST /invocations request with the given message.
+
+    When ``session_id`` is provided it is forwarded as the ``agent_session_id``
+    query parameter so the server reuses the same conversation session.
+    """
+    payload: dict[str, Any] = {"message": message, "stream": stream}
+    params = {"agent_session_id": session_id} if session_id is not None else None
+    transport = httpx.ASGITransport(app=server)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        return await client.post("/invocations", json=payload, params=params, timeout=120)
+
+
+# ---------------------------------------------------------------------------
+# Tests — basic text input
+# ---------------------------------------------------------------------------
+
+
+class TestBasicText:
+    """Simple text-in / text-out round trips."""
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_simple_text_non_streaming(self, server: InvocationsHostServer) -> None:
+        """Non-streaming: send a message and get the agent's text answer."""
+        resp = await _post_invocation(server, message="Say hello in exactly three words.", stream=False)
+
+        assert resp.status_code == 200
+        assert len(resp.text) > 0
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_simple_text_streaming(self, server: InvocationsHostServer) -> None:
+        """Streaming: send a message and receive text chunks as an event stream."""
+        resp = await _post_invocation(server, message="Say hello in exactly three words.", stream=True)
+
+        assert resp.status_code == 200
+        assert "text/event-stream" in resp.headers["content-type"]
+        assert len(resp.text) > 0
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_missing_message_returns_400(self, server: InvocationsHostServer) -> None:
+        """A request without a ``message`` field is rejected with a 400."""
+        transport = httpx.ASGITransport(app=server)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.post("/invocations", json={"stream": False}, timeout=120)
+
+        assert resp.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Tests — multi-turn conversations
+# ---------------------------------------------------------------------------
+
+
+class TestMultiTurn:
+    """Multi-round conversations using a shared agent_session_id."""
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_two_turn_conversation(self, server: InvocationsHostServer) -> None:
+        """Turn 1 establishes context; turn 2 recalls it via the same session."""
+        session_id = "int-test-session-two-turn"
+
+        resp1 = await _post_invocation(
+            server,
+            message="My favorite color is blue. Remember that.",
+            stream=False,
+            session_id=session_id,
+        )
+        assert resp1.status_code == 200
+
+        resp2 = await _post_invocation(
+            server,
+            message="What is my favorite color? Answer with a single word.",
+            stream=False,
+            session_id=session_id,
+        )
+        assert resp2.status_code == 200
+        assert "blue" in resp2.text.lower()
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_multi_turn_streaming(self, server: InvocationsHostServer) -> None:
+        """Multi-turn conversation with streaming on the second turn."""
+        session_id = "int-test-session-stream"
+
+        resp1 = await _post_invocation(
+            server,
+            message="My favorite number is 42.",
+            stream=False,
+            session_id=session_id,
+        )
+        assert resp1.status_code == 200
+
+        resp2 = await _post_invocation(
+            server,
+            message="What is my favorite number?",
+            stream=True,
+            session_id=session_id,
+        )
+        assert resp2.status_code == 200
+        assert "text/event-stream" in resp2.headers["content-type"]
+        assert "42" in resp2.text
+
+
+# ---------------------------------------------------------------------------
+# Tests — tool calling
+# ---------------------------------------------------------------------------
+
+
+class TestToolCalling:
+    """Tests that verify function-tool round trips through the hosting layer."""
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_tool_call_non_streaming(self, server_with_tools: InvocationsHostServer) -> None:
+        """Agent invokes a tool and returns a final answer (non-streaming)."""
+        resp = await _post_invocation(
+            server_with_tools,
+            message="What is the weather in Seattle?",
+            stream=False,
+        )
+
+        assert resp.status_code == 200
+        assert "72" in resp.text
+
+    @pytest.mark.flaky
+    @pytest.mark.integration
+    @skip_if_foundry_hosting_integration_tests_disabled
+    async def test_tool_call_streaming(self, server_with_tools: InvocationsHostServer) -> None:
+        """Agent invokes a tool and streams a final answer."""
+        resp = await _post_invocation(
+            server_with_tools,
+            message="What is the weather in Seattle?",
+            stream=True,
+        )
+
+        assert resp.status_code == 200
+        assert "text/event-stream" in resp.headers["content-type"]
+        assert "72" in resp.text

python/packages/foundry_hosting/tests/test_toolbox.py

@@ -4,7 +4,9 @@
 
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from collections.abc import Callable
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 from typing import cast
 from unittest.mock import AsyncMock
 
@@ -214,11 +216,11 @@ async def test_skills_source_uses_connected_session(monkeypatch: pytest.MonkeyPa
     sentinel_session = object()
     toolbox.session = sentinel_session  # type: ignore
 
-    captured: dict[str, object] = {}
+    captured: dict[str, Callable[[], object]] = {}
 
     class _StubSkillsSource:
-        def __init__(self, *, client: object) -> None:
-            captured["client"] = client
+        def __init__(self, *, session_provider: Callable[[], object]) -> None:
+            captured["session_provider"] = session_provider
 
         async def get_skills(self, context: SkillsSourceContext) -> list[str]:
             return ["skill-a"]
@@ -228,4 +230,105 @@ async def test_skills_source_uses_connected_session(monkeypatch: pytest.MonkeyPa
     result = await _FoundryToolboxSkillsSource(toolbox).get_skills(_source_context())
 
     assert result == ["skill-a"]
-    assert captured["client"] is sentinel_session
+    # The source hands MCPSkillsSource a provider (not a fixed session) that resolves
+    # the toolbox's current session, so it survives a reconnect that swaps it.
+    provider = captured["session_provider"]
+    assert provider() is sentinel_session
+    new_session = object()
+    toolbox.session = new_session  # type: ignore
+    assert provider() is new_session
+
+
+async def test_skills_source_requires_connection_via_provider() -> None:
+    toolbox = FoundryToolbox(
+        _FakeCredential(),  # type: ignore
+        url="https://h/toolboxes/tb/mcp",
+    )
+    toolbox.session = object()  # type: ignore
+    source = _FoundryToolboxSkillsSource(toolbox)
+    # Discovery captures the bound provider; a later reconnect gap (session is None)
+    # surfaces the same clear error when the provider is resolved.
+    toolbox.session = None  # type: ignore
+    with pytest.raises(RuntimeError, match="not connected"):
+        source._require_session()  # pyright: ignore[reportPrivateUsage]
+
+
+class _FakeSkill:
+    """Minimal stand-in for a :class:`~agent_framework.Skill` for caching tests."""
+
+    def __init__(self, name: str) -> None:
+        self.frontmatter = SimpleNamespace(name=name)
+
+
+def _patch_counting_mcp_source(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    """Patch ``MCPSkillsSource`` with a stub that counts index reads.
+
+    Returns a single-element list whose value tracks how many times
+    ``get_skills`` (i.e. a ``skill://index.json`` read) has been invoked.
+    """
+    read_count = [0]
+
+    class _CountingSkillsSource:
+        def __init__(self, *, session_provider: object) -> None:
+            self._session_provider = session_provider
+
+        async def get_skills(self, context: SkillsSourceContext) -> list[_FakeSkill]:
+            read_count[0] += 1
+            return [_FakeSkill("skill-a")]
+
+    monkeypatch.setattr("agent_framework_foundry_hosting._toolbox.MCPSkillsSource", _CountingSkillsSource)
+    return read_count
+
+
+async def test_as_skills_provider_caches_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    toolbox = FoundryToolbox(
+        _FakeCredential(),  # type: ignore
+        url="https://h/toolboxes/tb/mcp",
+    )
+    toolbox.session = object()  # type: ignore
+    read_count = _patch_counting_mcp_source(monkeypatch)
+
+    provider = toolbox.as_skills_provider()
+    context = _source_context()
+    for _ in range(3):
+        await provider._source.get_skills(context)  # pyright: ignore[reportPrivateUsage]
+
+    # By default the toolbox index is read once and reused across agent runs.
+    assert read_count[0] == 1
+
+
+async def test_as_skills_provider_disable_caching_rereads_every_run(monkeypatch: pytest.MonkeyPatch) -> None:
+    toolbox = FoundryToolbox(
+        _FakeCredential(),  # type: ignore
+        url="https://h/toolboxes/tb/mcp",
+    )
+    toolbox.session = object()  # type: ignore
+    read_count = _patch_counting_mcp_source(monkeypatch)
+
+    provider = toolbox.as_skills_provider(disable_caching=True)
+    context = _source_context()
+    for _ in range(3):
+        await provider._source.get_skills(context)  # pyright: ignore[reportPrivateUsage]
+
+    # With caching disabled the index is re-read on every agent run.
+    assert read_count[0] == 3
+
+
+async def test_as_skills_provider_cache_refresh_interval_rereads_after_staleness(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    toolbox = FoundryToolbox(
+        _FakeCredential(),  # type: ignore
+        url="https://h/toolboxes/tb/mcp",
+    )
+    toolbox.session = object()  # type: ignore
+    read_count = _patch_counting_mcp_source(monkeypatch)
+
+    # A zero interval makes every cached result immediately stale, so each run
+    # re-reads the index -- proving cache_refresh_interval is wired through.
+    provider = toolbox.as_skills_provider(cache_refresh_interval=timedelta(0))
+    context = _source_context()
+    for _ in range(3):
+        await provider._source.get_skills(context)  # pyright: ignore[reportPrivateUsage]
+
+    assert read_count[0] == 3

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

@@ -29,7 +29,7 @@ This directory contains samples that demonstrate how to use hosted [Agent Framew
 
 | # | Sample | Description |
 |---|--------|-------------|
-| 1 | [Basic](invocations/01_basic/) | A minimal agent demonstrating session state management via `agent_session_id` in URL params/response headers. |
+| 1 | [Basic](invocations/01_basic/) | A minimal agent demonstrating basic request/response using the invocations protocol. |
 | 2 | [Break Glass](invocations/02_break_glass/) | An agent demonstrating a "break glass" scenario where customizations of the API behaviors are needed, allowing for more direct control over how requests and responses are handled by the hosting layer. |
 
 ## Running the Agent Host Locally

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

@@ -28,6 +28,12 @@ Send a POST request to the server with a JSON body containing a "message" field
 curl -X POST http://localhost:8088/invocations -i -H "Content-Type: application/json" -d '{"message": "Hi"}'
 ```
 
+Or with streaming:
+
+```bash
+curl -X POST http://localhost:8088/invocations -i -H "Content-Type: application/json" -d '{"message": "Hi", "stream": true}'
+```
+
 The server will respond with a JSON object containing the response text. The `-i` flag in the `curl` command includes the HTTP response headers in the output, which includes the session ID that can be used for multi-turn conversations. Here is an example of the response:
 
 ```
@@ -40,7 +46,7 @@ x-platform-server: azure-ai-agentserver-core/2.0.0a20260410006 (python/3.12)
 date: Fri, 17 Apr 2026 23:46:44 GMT
 server: hypercorn-h11
 
-{"response":"Hi! How can I help?"}
+Hi! How can I help?
 ```
 
 ### Multi-turn conversation

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

@@ -40,7 +40,7 @@ x-platform-server: azure-ai-agentserver-core/2.0.0a20260410006 (python/3.12)
 date: Fri, 17 Apr 2026 23:46:44 GMT
 server: hypercorn-h11
 
-{"response":"Hi! How can I help?"}
+Hi! How can I help?
 ```
 
 ### Multi-turn conversation

python/samples/04-hosting/foundry-hosted-agents/invocations/02_break_glass/main.py

@@ -5,11 +5,12 @@ from collections.abc import AsyncGenerator
 
 from agent_framework import Agent, AgentSession
 from agent_framework.foundry import FoundryChatClient
+from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.invocations import InvocationAgentServerHost
 from azure.identity import DefaultAzureCredential
 from dotenv import load_dotenv
 from starlette.requests import Request
-from starlette.responses import JSONResponse, Response, StreamingResponse
+from starlette.responses import Response, StreamingResponse
 
 # Load environment variables from .env file
 load_dotenv()
@@ -38,12 +39,31 @@ agent = Agent(
 app = InvocationAgentServerHost()
 
 
+def get_session_partition_key() -> str:
+    """Get the partition key for the current request.
+
+    A partition key is made up of the session ID and user ID. If the request is not
+    from a hosted environment, the partition key will be just the session ID. In the
+    Foundry hosted environment, the partition key is used to maintain isolation between
+    different sessions and users, such that one user cannot access another user's sessions.
+
+    Returns:
+        The partition key for the current request.
+    """
+    context = get_request_context()
+    if context.session_id is None:
+        raise RuntimeError(
+            "The request context is missing session_id. Please ensure that the request is a valid request."
+        )
+    if context.user_id is not None:
+        return f"{context.session_id}:{context.user_id}"
+    return context.session_id
+
+
 @app.invoke_handler
 async def handle_invoke(request: Request):
     """Handle streaming multi-turn chat with Azure OpenAI via SSE."""
     data = await request.json()
-    session_id = request.state.session_id
-
     stream = data.get("stream", False)
     user_message = data.get("message", None)
     if user_message is None:
@@ -52,6 +72,7 @@ async def handle_invoke(request: Request):
             return StreamingResponse(content=error, status_code=400)
         return Response(content=error, status_code=400)
 
+    session_id = get_session_partition_key()
     session = _sessions.setdefault(session_id, AgentSession(session_id=session_id))
 
     if stream:
@@ -67,7 +88,7 @@ async def handle_invoke(request: Request):
         )
 
     response = await agent.run([user_message], session=session, stream=stream)
-    return JSONResponse({"response": response.text})
+    return Response(content=response.text)
 
 
 if __name__ == "__main__":