From e67aef89e0005ffabc12d832aa7972237ed3a388 Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:49:39 -0700 Subject: [PATCH 1/7] Python: Reject caller-supplied tool execution context in Responses requests ResponsesHostServer now rejects additional_function_arguments supplied at the top level or in extra_body, including on approval resume, so callers cannot replace host-configured tool context. Developer defaults and prepare_options hooks can still supply trusted values. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/foundry_hosting/README.md | 6 +- .../_request.py | 23 ++- .../foundry_hosting/tests/test_request.py | 34 +++++ .../foundry_hosting/tests/test_responses.py | 139 ++++++++++++++++++ 4 files changed, 199 insertions(+), 3 deletions(-) diff --git a/python/packages/foundry_hosting/README.md b/python/packages/foundry_hosting/README.md index 3a352eba761..24344de793e 100644 --- a/python/packages/foundry_hosting/README.md +++ b/python/packages/foundry_hosting/README.md @@ -238,7 +238,11 @@ Native CreateResponse generation fields become MAF runtime options (notably `max **last**. A sync or async `prepare_options(request: HostedResponseRequest, options: dict)` hook can remove or replace *caller* options before `Agent.run`; removed values fall back to the developer's unchanged agent defaults. Hosting filters caller platform IDs and private continuation/storage controls from model options and rejects attempts to -reintroduce them through the hook. Nested `extra_body` transport overrides are rejected for both caller +reintroduce them through the hook. Callers cannot supply `additional_function_arguments` (the hidden tool-execution +context that reaches `FunctionInvocationContext.kwargs`), at the top level or in `extra_body`, on any request including +an approval resume: the request fails before any tool runs and pending approvals remain usable. Set trusted values in +the agent's `default_options` or return them from a `prepare_options` hook (for example, derived from +`request.scope`). Nested `extra_body` transport overrides are rejected for both caller input and developer hooks, because they could override the host's `store=False` after the OpenAI SDK merges the body. Developer defaults also cannot use this transport channel for host-controlled fields on explicit history modes or unstored requests. A custom agent cannot accept MAF runtime options: choose diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py index 335658f276f..3056dff375e 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py @@ -5,7 +5,7 @@ from __future__ import annotations import inspect -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence from copy import deepcopy from dataclasses import dataclass, field from types import MappingProxyType @@ -116,14 +116,32 @@ def __post_init__(self) -> None: }) _OPTION_NAMES = {"max_output_tokens": "max_tokens", "parallel_tool_calls": "allow_multiple_tool_calls"} _NATIVE_FIELDS = frozenset(CreateResponse.__annotations__) +# Trusted tool-execution context that callers must not supply. Unlike ``_HOST_CONTROLLED_FIELDS``, a developer +# ``prepare_options`` hook may still set it from trusted host state. +_CALLER_REJECTED_FIELDS = frozenset({"additional_function_arguments"}) + + +def _reject_caller_controls(names: Iterable[str]) -> None: + rejected = _CALLER_REJECTED_FIELDS.intersection(names) + if rejected: + raise ValueError( + f"Responses requests cannot set agent-controlled fields: {', '.join(sorted(rejected))}. " + "Configure trusted tool context in the agent's default options or a prepare_options hook." + ) def response_run_options(request: CreateResponse) -> dict[str, Any]: - """Translate native generation fields, with flattened extra-body values winning collisions.""" + """Translate native generation fields, with flattened extra-body values winning collisions. + + Raises: + ValueError: If the caller supplies agent-controlled tool-execution context, at the top level or in + ``extra_body``, so it can never replace trusted values configured by the developer. + """ native: dict[str, Any] = {} extra: dict[str, Any] = {} nested_extra: dict[str, Any] = {} for name, value in request.items(): + _reject_caller_controls((name,)) if value is None or (name == "model" and value == ""): continue if name == "extra_body": @@ -133,6 +151,7 @@ def response_run_options(request: CreateResponse) -> dict[str, Any]: raise TypeError("extra_body must be a mapping of model options.") if "extra_body" in value: raise ValueError("Nested extra_body is not supported; flatten provider options instead.") + _reject_caller_controls(cast(Mapping[str, Any], value)) nested_extra.update(cast(Mapping[str, Any], value)) elif name in _HOST_CONTROLLED_FIELDS: continue diff --git a/python/packages/foundry_hosting/tests/test_request.py b/python/packages/foundry_hosting/tests/test_request.py index 6496e7e8b6f..489bbda6a9f 100644 --- a/python/packages/foundry_hosting/tests/test_request.py +++ b/python/packages/foundry_hosting/tests/test_request.py @@ -75,6 +75,40 @@ def test_nested_extra_body_is_also_overlaid_last_without_reserved_fields() -> No response_run_options(cast(CreateResponse, {"extra_body": {"extra_body": {"store": True}}})) +@pytest.mark.parametrize( + "payload", + [ + {"additional_function_arguments": {"tenant_id": "forged"}}, + {"extra_body": {"additional_function_arguments": {"tenant_id": "forged"}}}, + {"additional_function_arguments": None}, + {"extra_body": {"additional_function_arguments": None}}, + ], +) +def test_caller_cannot_supply_tool_execution_context(payload: dict[str, Any]) -> None: + with pytest.raises(ValueError, match="agent-controlled fields: additional_function_arguments"): + response_run_options(cast(CreateResponse, {"input": "hello", "temperature": 0.2, **payload})) + + +def test_requests_without_tool_context_keep_other_options() -> None: + request = cast(CreateResponse, {"input": "hello", "extra_body": {"slogan_style": "retro"}, "temperature": 0.2}) + assert response_run_options(request) == {"temperature": 0.2, "slogan_style": "retro"} + + +async def test_hook_can_supply_trusted_tool_context() -> None: + context = MagicMock(spec=ResponseContext) + context.response_id = "outer-id" + context.conversation_id = None + scope = FoundryRequestScope(session_id="sandbox", user_id="user", call_id="call", is_hosted=True) + request = HostedResponseRequest(cast(CreateResponse, {"input": "hello"}), context, scope, {}) + + def inject(view: HostedResponseRequest, options: dict[str, Any]) -> dict[str, Any]: + return {**options, "additional_function_arguments": {"tenant_id": view.scope.user_id}} + + await prepare_response_options(request, inject) + validate_request_options(request.options) + assert request.options["additional_function_arguments"] == {"tenant_id": "user"} + + async def test_hook_uses_a_request_copy_and_does_not_mutate_defaults() -> None: payload = cast(CreateResponse, {"input": "hello", "metadata": {"source": "caller"}}) context = MagicMock(spec=ResponseContext) diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index dc9b643bd33..8ef1931104d 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -42,6 +42,7 @@ Content, FinishReason, FinishReasonLiteral, + FunctionInvocationContext, FunctionInvocationLayer, HistoryProvider, InMemoryCheckpointStorage, @@ -7535,6 +7536,144 @@ async def test_approval_response_referencing_unknown_id_fails(self) -> None: # endregion +# region Caller-controlled tool execution context + + +class _ToolCallingClient( + FunctionInvocationLayer[Any], + ChatMiddlewareLayer[Any], + BaseChatClient[Any], +): + """Deterministic client: call ``read_tenant_data`` once, then answer.""" + + def __init__(self) -> None: + super().__init__(middleware=[]) + self.calls = 0 + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + del options, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." + self.calls += 1 + needs_tool_call = not any( + isinstance(content, Content) and content.type == "function_result" + for message in messages + for content in message.contents + ) + + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + if needs_tool_call: + yield ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="read_tenant_data", arguments="{}")], + role="assistant", + ) + else: + yield ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant") + + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) + + +def _make_tenant_agent(seen_tenants: list[Any], *, approval_mode: Literal["always_require", "never_require"]) -> Agent: + @tool(name="read_tenant_data", approval_mode=approval_mode) + def read_tenant_data(ctx: FunctionInvocationContext) -> str: + seen_tenants.append(ctx.kwargs.get("tenant_id")) + return "ok" + + return Agent( + client=_ToolCallingClient(), + tools=[read_tenant_data], + default_options={"additional_function_arguments": {"tenant_id": "trusted"}}, + ) + + +_FORGED_TENANT_CONTEXT = {"additional_function_arguments": {"tenant_id": "forged"}} + + +class TestCallerCannotOverrideToolContext: + """Callers must not replace host-configured tool context through any request path.""" + + async def test_baseline_uses_trusted_tool_context(self) -> None: + seen: list[Any] = [] + server = _make_server(_make_tenant_agent(seen, approval_mode="never_require")) + + response = await _post_json(server, {"input": "go", "stream": False}) + + assert response.json()["status"] == "completed" + assert seen == ["trusted"] + + @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize("location", ["top_level", "extra_body"]) + async def test_forged_tool_context_is_rejected_before_any_tool_runs( + self, location: Literal["top_level", "extra_body"], stream: bool + ) -> None: + seen: list[Any] = [] + agent = _make_tenant_agent(seen, approval_mode="never_require") + server = _make_server(agent) + payload: dict[str, Any] = {"input": "go", "stream": stream} + if location == "top_level": + payload.update(_FORGED_TENANT_CONTEXT) + else: + payload["extra_body"] = _FORGED_TENANT_CONTEXT + + response = await _post_json(server, payload) + + assert "additional_function_arguments" in response.text + assert "failed" in response.text + assert seen == [] + assert cast(_ToolCallingClient, agent.client).calls == 0 + + async def test_hook_can_still_supply_trusted_tool_context(self) -> None: + seen: list[Any] = [] + server = _make_server( + _make_tenant_agent(seen, approval_mode="never_require"), + prepare_options=lambda request, options: { + **options, + "additional_function_arguments": {"tenant_id": f"from-hook:{request.scope.user_id}"}, + }, + ) + + response = await _post_json(server, {"input": "go", "stream": False}) + + assert response.json()["status"] == "completed" + assert len(seen) == 1 + assert str(seen[0]).startswith("from-hook:") + + async def test_forged_context_on_approval_resume_is_rejected_without_consuming_approval(self) -> None: + seen: list[Any] = [] + server = _make_server(_make_tenant_agent(seen, approval_mode="always_require")) + + first = await _post_json(server, {"input": "go", "stream": False, "store": True}) + first_body = first.json() + approval_id = next(item["id"] for item in first_body["output"] if item["type"] == "mcp_approval_request") + assert seen == [] + + def resume(**extra: Any) -> dict[str, Any]: + return { + "input": [{"type": "mcp_approval_response", "approval_request_id": approval_id, "approve": True}], + "previous_response_id": first_body["id"], + "stream": False, + "store": True, + **extra, + } + + forged = await _post_json(server, resume(**_FORGED_TENANT_CONTEXT)) + assert "additional_function_arguments" in forged.text + assert seen == [] + + legitimate = await _post_json(server, resume()) + assert legitimate.json()["status"] == "completed" + assert seen == ["trusted"] + + +# endregion + + # region Checkpoint context validation From 77cc450b4bd9a23a6ad4f7b96e367a840ecb1d56 Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:01:08 -0700 Subject: [PATCH 2/7] Move caller tool-context tests to tests/foundry_hosting with a helper module Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../test_caller_tool_context.py | 147 ++++++++++++++++++ .../foundry_hosting/tool_context_helpers.py | 80 ++++++++++ .../foundry_hosting/tests/test_request.py | 34 ---- .../foundry_hosting/tests/test_responses.py | 139 ----------------- 4 files changed, 227 insertions(+), 173 deletions(-) create mode 100644 python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py create mode 100644 python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py new file mode 100644 index 00000000000..df91fefa297 --- /dev/null +++ b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py @@ -0,0 +1,147 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Callers must not supply or replace trusted tool-execution context on Responses requests.""" + +from __future__ import annotations + +from typing import Any, Literal, cast +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest +from azure.ai.agentserver.responses import InMemoryResponseProvider, ResponseContext +from azure.ai.agentserver.responses.models import CreateResponse +from tool_context_helpers import ( # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] + FORGED_TENANT_CONTEXT, + TRUSTED_TENANT_ID, + ToolCallingClient, + make_tenant_agent, +) + +from agent_framework_foundry_hosting import HostedResponseRequest, ResponsesHostServer +from agent_framework_foundry_hosting._request import ( + prepare_response_options, + response_run_options, + validate_request_options, +) +from agent_framework_foundry_hosting._scope import FoundryRequestScope + + +@pytest.mark.parametrize( + "payload", + [ + {"additional_function_arguments": {"tenant_id": "forged"}}, + {"extra_body": {"additional_function_arguments": {"tenant_id": "forged"}}}, + {"additional_function_arguments": None}, + {"extra_body": {"additional_function_arguments": None}}, + ], +) +def test_caller_cannot_supply_tool_execution_context(payload: dict[str, Any]) -> None: + with pytest.raises(ValueError, match="agent-controlled fields: additional_function_arguments"): + response_run_options(cast(CreateResponse, {"input": "hello", "temperature": 0.2, **payload})) + + +def test_requests_without_tool_context_keep_other_options() -> None: + request = cast(CreateResponse, {"input": "hello", "extra_body": {"slogan_style": "retro"}, "temperature": 0.2}) + assert response_run_options(request) == {"temperature": 0.2, "slogan_style": "retro"} + + +async def test_hook_can_supply_trusted_tool_context() -> None: + context = MagicMock(spec=ResponseContext) + context.response_id = "outer-id" + context.conversation_id = None + context.get_input_text = AsyncMock(return_value="hello") + scope = FoundryRequestScope(session_id="sandbox", user_id="user", call_id="call", is_hosted=True) + request = HostedResponseRequest(cast(CreateResponse, {"input": "hello"}), context, scope, {}) + + def inject(view: HostedResponseRequest, options: dict[str, Any]) -> dict[str, Any]: + return {**options, "additional_function_arguments": {"tenant_id": view.scope.user_id}} + + await prepare_response_options(request, inject) + validate_request_options(request.options) + assert request.options["additional_function_arguments"] == {"tenant_id": "user"} + + +async def _post_json(server: ResponsesHostServer, payload: dict[str, Any]) -> httpx.Response: + transport = httpx.ASGITransport(app=server) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + return await client.post("/responses", json=payload) + + +def _make_server(agent: Any, **kwargs: Any) -> ResponsesHostServer: + return ResponsesHostServer(agent, response_store=InMemoryResponseProvider(), **kwargs) + + +@pytest.mark.xdist_group("session_isolation") +class TestCallerCannotOverrideToolContext: + async def test_baseline_uses_trusted_tool_context(self) -> None: + seen: list[Any] = [] + server = _make_server(make_tenant_agent(seen, approval_mode="never_require")) + + response = await _post_json(server, {"input": "go", "stream": False}) + + assert response.json()["status"] == "completed" + assert seen == [TRUSTED_TENANT_ID] + + @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize("location", ["top_level", "extra_body"]) + async def test_forged_tool_context_is_rejected_before_any_tool_runs( + self, location: Literal["top_level", "extra_body"], stream: bool + ) -> None: + seen: list[Any] = [] + agent = make_tenant_agent(seen, approval_mode="never_require") + server = _make_server(agent) + payload: dict[str, Any] = {"input": "go", "stream": stream} + if location == "top_level": + payload.update(FORGED_TENANT_CONTEXT) + else: + payload["extra_body"] = FORGED_TENANT_CONTEXT + + response = await _post_json(server, payload) + + assert "additional_function_arguments" in response.text + assert "failed" in response.text + assert seen == [] + assert cast(ToolCallingClient, agent.client).calls == 0 + + async def test_hook_can_still_supply_trusted_tool_context(self) -> None: + seen: list[Any] = [] + server = _make_server( + make_tenant_agent(seen, approval_mode="never_require"), + prepare_options=lambda request, options: { + **options, + "additional_function_arguments": {"tenant_id": f"from-hook:{request.scope.user_id}"}, + }, + ) + + response = await _post_json(server, {"input": "go", "stream": False}) + + assert response.json()["status"] == "completed" + assert len(seen) == 1 + assert str(seen[0]).startswith("from-hook:") + + async def test_forged_context_on_approval_resume_is_rejected_without_consuming_approval(self) -> None: + seen: list[Any] = [] + server = _make_server(make_tenant_agent(seen, approval_mode="always_require")) + + first = await _post_json(server, {"input": "go", "stream": False, "store": True}) + first_body = first.json() + approval_id = next(item["id"] for item in first_body["output"] if item["type"] == "mcp_approval_request") + assert seen == [] + + def resume(**extra: Any) -> dict[str, Any]: + return { + "input": [{"type": "mcp_approval_response", "approval_request_id": approval_id, "approve": True}], + "previous_response_id": first_body["id"], + "stream": False, + "store": True, + **extra, + } + + forged = await _post_json(server, resume(**FORGED_TENANT_CONTEXT)) + assert "additional_function_arguments" in forged.text + assert seen == [] + + legitimate = await _post_json(server, resume()) + assert legitimate.json()["status"] == "completed" + assert seen == [TRUSTED_TENANT_ID] diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py new file mode 100644 index 00000000000..6135b679615 --- /dev/null +++ b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py @@ -0,0 +1,80 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Deterministic agent helpers for hosted tool-execution-context tests.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Awaitable, Mapping, Sequence +from typing import Any, Literal + +from agent_framework import ( + Agent, + BaseChatClient, + ChatMiddlewareLayer, + ChatResponse, + ChatResponseUpdate, + Content, + FunctionInvocationContext, + FunctionInvocationLayer, + Message, + ResponseStream, + tool, +) + +TRUSTED_TENANT_ID = "trusted" +FORGED_TENANT_CONTEXT: dict[str, Any] = {"additional_function_arguments": {"tenant_id": "forged"}} + + +class ToolCallingClient( + FunctionInvocationLayer[Any], + ChatMiddlewareLayer[Any], + BaseChatClient[Any], +): + """Call ``read_tenant_data`` once, then answer.""" + + def __init__(self) -> None: + super().__init__(middleware=[]) + self.calls = 0 + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + del options, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." + self.calls += 1 + needs_tool_call = not any( + content.type == "function_result" for message in messages for content in message.contents + ) + + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + if needs_tool_call: + yield ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="read_tenant_data", arguments="{}")], + role="assistant", + ) + else: + yield ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant") + + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) + + +def make_tenant_agent( + seen_tenants: list[Any], *, approval_mode: Literal["always_require", "never_require"] +) -> Agent[Any]: + """Create an agent whose tool records the tenant it receives from hidden runtime context.""" + + @tool(name="read_tenant_data", approval_mode=approval_mode) + def read_tenant_data(ctx: FunctionInvocationContext) -> str: + seen_tenants.append(ctx.kwargs.get("tenant_id")) + return "ok" + + return Agent( + client=ToolCallingClient(), + tools=[read_tenant_data], + default_options={"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}, + ) diff --git a/python/packages/foundry_hosting/tests/test_request.py b/python/packages/foundry_hosting/tests/test_request.py index 489bbda6a9f..6496e7e8b6f 100644 --- a/python/packages/foundry_hosting/tests/test_request.py +++ b/python/packages/foundry_hosting/tests/test_request.py @@ -75,40 +75,6 @@ def test_nested_extra_body_is_also_overlaid_last_without_reserved_fields() -> No response_run_options(cast(CreateResponse, {"extra_body": {"extra_body": {"store": True}}})) -@pytest.mark.parametrize( - "payload", - [ - {"additional_function_arguments": {"tenant_id": "forged"}}, - {"extra_body": {"additional_function_arguments": {"tenant_id": "forged"}}}, - {"additional_function_arguments": None}, - {"extra_body": {"additional_function_arguments": None}}, - ], -) -def test_caller_cannot_supply_tool_execution_context(payload: dict[str, Any]) -> None: - with pytest.raises(ValueError, match="agent-controlled fields: additional_function_arguments"): - response_run_options(cast(CreateResponse, {"input": "hello", "temperature": 0.2, **payload})) - - -def test_requests_without_tool_context_keep_other_options() -> None: - request = cast(CreateResponse, {"input": "hello", "extra_body": {"slogan_style": "retro"}, "temperature": 0.2}) - assert response_run_options(request) == {"temperature": 0.2, "slogan_style": "retro"} - - -async def test_hook_can_supply_trusted_tool_context() -> None: - context = MagicMock(spec=ResponseContext) - context.response_id = "outer-id" - context.conversation_id = None - scope = FoundryRequestScope(session_id="sandbox", user_id="user", call_id="call", is_hosted=True) - request = HostedResponseRequest(cast(CreateResponse, {"input": "hello"}), context, scope, {}) - - def inject(view: HostedResponseRequest, options: dict[str, Any]) -> dict[str, Any]: - return {**options, "additional_function_arguments": {"tenant_id": view.scope.user_id}} - - await prepare_response_options(request, inject) - validate_request_options(request.options) - assert request.options["additional_function_arguments"] == {"tenant_id": "user"} - - async def test_hook_uses_a_request_copy_and_does_not_mutate_defaults() -> None: payload = cast(CreateResponse, {"input": "hello", "metadata": {"source": "caller"}}) context = MagicMock(spec=ResponseContext) diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 8ef1931104d..dc9b643bd33 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -42,7 +42,6 @@ Content, FinishReason, FinishReasonLiteral, - FunctionInvocationContext, FunctionInvocationLayer, HistoryProvider, InMemoryCheckpointStorage, @@ -7536,144 +7535,6 @@ async def test_approval_response_referencing_unknown_id_fails(self) -> None: # endregion -# region Caller-controlled tool execution context - - -class _ToolCallingClient( - FunctionInvocationLayer[Any], - ChatMiddlewareLayer[Any], - BaseChatClient[Any], -): - """Deterministic client: call ``read_tenant_data`` once, then answer.""" - - def __init__(self) -> None: - super().__init__(middleware=[]) - self.calls = 0 - - def _inner_get_response( - self, - *, - messages: Sequence[Message], - stream: bool, - options: Mapping[str, Any], - **kwargs: Any, - ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: - del options, kwargs - assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." - self.calls += 1 - needs_tool_call = not any( - isinstance(content, Content) and content.type == "function_result" - for message in messages - for content in message.contents - ) - - async def stream_response() -> AsyncIterator[ChatResponseUpdate]: - if needs_tool_call: - yield ChatResponseUpdate( - contents=[Content.from_function_call(call_id="call_1", name="read_tenant_data", arguments="{}")], - role="assistant", - ) - else: - yield ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant") - - return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) - - -def _make_tenant_agent(seen_tenants: list[Any], *, approval_mode: Literal["always_require", "never_require"]) -> Agent: - @tool(name="read_tenant_data", approval_mode=approval_mode) - def read_tenant_data(ctx: FunctionInvocationContext) -> str: - seen_tenants.append(ctx.kwargs.get("tenant_id")) - return "ok" - - return Agent( - client=_ToolCallingClient(), - tools=[read_tenant_data], - default_options={"additional_function_arguments": {"tenant_id": "trusted"}}, - ) - - -_FORGED_TENANT_CONTEXT = {"additional_function_arguments": {"tenant_id": "forged"}} - - -class TestCallerCannotOverrideToolContext: - """Callers must not replace host-configured tool context through any request path.""" - - async def test_baseline_uses_trusted_tool_context(self) -> None: - seen: list[Any] = [] - server = _make_server(_make_tenant_agent(seen, approval_mode="never_require")) - - response = await _post_json(server, {"input": "go", "stream": False}) - - assert response.json()["status"] == "completed" - assert seen == ["trusted"] - - @pytest.mark.parametrize("stream", [False, True]) - @pytest.mark.parametrize("location", ["top_level", "extra_body"]) - async def test_forged_tool_context_is_rejected_before_any_tool_runs( - self, location: Literal["top_level", "extra_body"], stream: bool - ) -> None: - seen: list[Any] = [] - agent = _make_tenant_agent(seen, approval_mode="never_require") - server = _make_server(agent) - payload: dict[str, Any] = {"input": "go", "stream": stream} - if location == "top_level": - payload.update(_FORGED_TENANT_CONTEXT) - else: - payload["extra_body"] = _FORGED_TENANT_CONTEXT - - response = await _post_json(server, payload) - - assert "additional_function_arguments" in response.text - assert "failed" in response.text - assert seen == [] - assert cast(_ToolCallingClient, agent.client).calls == 0 - - async def test_hook_can_still_supply_trusted_tool_context(self) -> None: - seen: list[Any] = [] - server = _make_server( - _make_tenant_agent(seen, approval_mode="never_require"), - prepare_options=lambda request, options: { - **options, - "additional_function_arguments": {"tenant_id": f"from-hook:{request.scope.user_id}"}, - }, - ) - - response = await _post_json(server, {"input": "go", "stream": False}) - - assert response.json()["status"] == "completed" - assert len(seen) == 1 - assert str(seen[0]).startswith("from-hook:") - - async def test_forged_context_on_approval_resume_is_rejected_without_consuming_approval(self) -> None: - seen: list[Any] = [] - server = _make_server(_make_tenant_agent(seen, approval_mode="always_require")) - - first = await _post_json(server, {"input": "go", "stream": False, "store": True}) - first_body = first.json() - approval_id = next(item["id"] for item in first_body["output"] if item["type"] == "mcp_approval_request") - assert seen == [] - - def resume(**extra: Any) -> dict[str, Any]: - return { - "input": [{"type": "mcp_approval_response", "approval_request_id": approval_id, "approve": True}], - "previous_response_id": first_body["id"], - "stream": False, - "store": True, - **extra, - } - - forged = await _post_json(server, resume(**_FORGED_TENANT_CONTEXT)) - assert "additional_function_arguments" in forged.text - assert seen == [] - - legitimate = await _post_json(server, resume()) - assert legitimate.json()["status"] == "completed" - assert seen == ["trusted"] - - -# endregion - - # region Checkpoint context validation From 5ace1879457dd72e87d78d55287fcb99dd28d130 Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:04:13 -0700 Subject: [PATCH 3/7] Apply caller tool-context rejection to legacy WorkflowAgent Responses requests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../_request.py | 21 ++++++++++--- .../_responses.py | 5 ++++ .../test_caller_tool_context.py | 30 +++++++++++++++++++ .../foundry_hosting/tool_context_helpers.py | 17 +++++++++++ 4 files changed, 69 insertions(+), 4 deletions(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py index 3056dff375e..277bca76344 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_request.py @@ -130,18 +130,32 @@ def _reject_caller_controls(names: Iterable[str]) -> None: ) +def reject_caller_tool_context(request: CreateResponse) -> None: + """Reject caller-supplied agent-controlled tool context, at the top level or in ``extra_body``. + + This applies to every hosted Responses request, whichever agent or workflow shape serves it, so callers can + never replace trusted values configured by the developer. + + Raises: + ValueError: If the request sets an agent-controlled field. + """ + _reject_caller_controls(cast(Mapping[str, Any], request)) + extra_body = request.get("extra_body") + if isinstance(extra_body, Mapping): + _reject_caller_controls(cast(Mapping[str, Any], extra_body)) + + def response_run_options(request: CreateResponse) -> dict[str, Any]: """Translate native generation fields, with flattened extra-body values winning collisions. Raises: - ValueError: If the caller supplies agent-controlled tool-execution context, at the top level or in - ``extra_body``, so it can never replace trusted values configured by the developer. + ValueError: If the caller supplies agent-controlled tool-execution context. """ + reject_caller_tool_context(request) native: dict[str, Any] = {} extra: dict[str, Any] = {} nested_extra: dict[str, Any] = {} for name, value in request.items(): - _reject_caller_controls((name,)) if value is None or (name == "model" and value == ""): continue if name == "extra_body": @@ -151,7 +165,6 @@ def response_run_options(request: CreateResponse) -> dict[str, Any]: raise TypeError("extra_body must be a mapping of model options.") if "extra_body" in value: raise ValueError("Nested extra_body is not supported; flatten provider options instead.") - _reject_caller_controls(cast(Mapping[str, Any], value)) nested_extra.update(cast(Mapping[str, Any], value)) elif name in _HOST_CONTROLLED_FIELDS: continue 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 09692723734..0fce25cb093 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -110,6 +110,7 @@ UnsupportedOptions, WorkflowTurn, prepare_response_options, + reject_caller_tool_context, response_run_options, validate_default_transport_options, validate_request_options, @@ -1154,6 +1155,10 @@ async def _handle_response( _initialize_agent_history(agent, configuration) if configuration.workflow: self._warn_legacy_workflow() + try: + reject_caller_tool_context(request) + except ValueError as exc: + raise _PublicResponseFailure(exc) from None hosted_request: HostedResponseRequest | None = None if configuration.workflow: if self._prepare_options is not None: diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py index df91fefa297..d0b74669027 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py @@ -16,6 +16,7 @@ TRUSTED_TENANT_ID, ToolCallingClient, make_tenant_agent, + make_tenant_workflow_agent, ) from agent_framework_foundry_hosting import HostedResponseRequest, ResponsesHostServer @@ -104,6 +105,35 @@ async def test_forged_tool_context_is_rejected_before_any_tool_runs( assert seen == [] assert cast(ToolCallingClient, agent.client).calls == 0 + @pytest.mark.filterwarnings("ignore:Hosting WorkflowAgent through agent=.*:DeprecationWarning") + @pytest.mark.parametrize("location", ["top_level", "extra_body"]) + async def test_forged_tool_context_is_rejected_for_legacy_workflow_agents( + self, location: Literal["top_level", "extra_body"] + ) -> None: + seen: list[Any] = [] + server = _make_server(make_tenant_workflow_agent(seen)) + payload: dict[str, Any] = {"input": "go", "stream": False} + if location == "top_level": + payload.update(FORGED_TENANT_CONTEXT) + else: + payload["extra_body"] = FORGED_TENANT_CONTEXT + + response = await _post_json(server, payload) + + assert "additional_function_arguments" in response.text + assert response.json()["status"] == "failed" + assert seen == [] + + @pytest.mark.filterwarnings("ignore:Hosting WorkflowAgent through agent=.*:DeprecationWarning") + async def test_legacy_workflow_agent_without_forged_context_still_runs(self) -> None: + seen: list[Any] = [] + server = _make_server(make_tenant_workflow_agent(seen)) + + response = await _post_json(server, {"input": "go", "stream": False}) + + assert response.json()["status"] == "completed" + assert seen == [TRUSTED_TENANT_ID] + async def test_hook_can_still_supply_trusted_tool_context(self) -> None: seen: list[Any] = [] server = _make_server( diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py index 6135b679615..f6356ea08d3 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py @@ -9,6 +9,7 @@ from agent_framework import ( Agent, + AgentExecutorRequest, BaseChatClient, ChatMiddlewareLayer, ChatResponse, @@ -18,6 +19,10 @@ FunctionInvocationLayer, Message, ResponseStream, + WorkflowAgent, + WorkflowBuilder, + WorkflowContext, + executor, tool, ) @@ -78,3 +83,15 @@ def read_tenant_data(ctx: FunctionInvocationContext) -> str: tools=[read_tenant_data], default_options={"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}, ) + + +def make_tenant_workflow_agent(seen_tenants: list[Any]) -> WorkflowAgent: + """Wrap the tenant agent in a legacy ``WorkflowAgent`` so the workflow hosting path can be exercised.""" + + @executor + async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None: + await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True)) + + inner = make_tenant_agent(seen_tenants, approval_mode="never_require") + workflow = WorkflowBuilder(start_executor=start).add_edge(start, inner).build() + return WorkflowAgent(workflow=workflow, name="Tenant Workflow Agent") From 07ac9b2ce651e272ca3b39dafedf164f644a98cb Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:17:46 -0700 Subject: [PATCH 4/7] Fix pyright error in tool-context test helper default_options Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../tests/foundry_hosting/tool_context_helpers.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py index f6356ea08d3..f842ab5e0d8 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import AsyncIterator, Awaitable, Mapping, Sequence -from typing import Any, Literal +from typing import Any, Literal, cast from agent_framework import ( Agent, @@ -81,7 +81,7 @@ def read_tenant_data(ctx: FunctionInvocationContext) -> str: return Agent( client=ToolCallingClient(), tools=[read_tenant_data], - default_options={"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}, + default_options=cast(Any, {"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}), ) From 86e90d1a4c185e94797c9a2a50b84d659d921396 Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:02:19 -0700 Subject: [PATCH 5/7] Make foundry_hosting test helpers importable under --import-mode=importlib Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../tests/foundry_hosting/conftest.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 python/packages/foundry_hosting/tests/foundry_hosting/conftest.py diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py b/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py new file mode 100644 index 00000000000..61818a7e62b --- /dev/null +++ b/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py @@ -0,0 +1,13 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Make helper modules in this directory importable by name under any pytest import mode.""" + +import sys +from pathlib import Path + + +def pytest_configure() -> None: + """Put this directory on sys.path; the workspace test sweep uses --import-mode=importlib.""" + test_dir = str(Path(__file__).resolve().parent) + if test_dir not in sys.path: + sys.path.insert(0, test_dir) From 307b7b9e969beb830cbd0a7bc4088c3c06b5109f Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:15:48 -0700 Subject: [PATCH 6/7] Replace sys.path test helper with fixtures to avoid shadowing other packages' conftest Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../tests/foundry_hosting/conftest.py | 109 ++++++++++++++++-- .../test_caller_tool_context.py | 41 +++---- .../foundry_hosting/tool_context_helpers.py | 97 ---------------- 3 files changed, 122 insertions(+), 125 deletions(-) delete mode 100644 python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py b/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py index 61818a7e62b..744e9505e86 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/conftest.py @@ -1,13 +1,106 @@ # Copyright (c) Microsoft. All rights reserved. -"""Make helper modules in this directory importable by name under any pytest import mode.""" +"""Fixtures for hosted tool-execution-context tests.""" -import sys -from pathlib import Path +from __future__ import annotations +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence +from typing import Any, Literal, cast -def pytest_configure() -> None: - """Put this directory on sys.path; the workspace test sweep uses --import-mode=importlib.""" - test_dir = str(Path(__file__).resolve().parent) - if test_dir not in sys.path: - sys.path.insert(0, test_dir) +import pytest +from agent_framework import ( + Agent, + AgentExecutorRequest, + BaseChatClient, + ChatMiddlewareLayer, + ChatResponse, + ChatResponseUpdate, + Content, + FunctionInvocationContext, + FunctionInvocationLayer, + Message, + ResponseStream, + WorkflowAgent, + WorkflowBuilder, + WorkflowContext, + executor, + tool, +) + +TRUSTED_TENANT_ID = "trusted" + +TenantAgentFactory = Callable[..., Agent[Any]] +TenantWorkflowAgentFactory = Callable[[list[Any]], WorkflowAgent] + + +class _ToolCallingClient( + FunctionInvocationLayer[Any], + ChatMiddlewareLayer[Any], + BaseChatClient[Any], +): + """Call ``read_tenant_data`` once, then answer.""" + + def __init__(self) -> None: + super().__init__(middleware=[]) + self.calls = 0 + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + del options, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." + self.calls += 1 + needs_tool_call = not any( + content.type == "function_result" for message in messages for content in message.contents + ) + + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + if needs_tool_call: + yield ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="read_tenant_data", arguments="{}")], + role="assistant", + ) + else: + yield ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant") + + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) + + +@pytest.fixture +def tenant_agent_factory() -> TenantAgentFactory: + """Create agents whose tool records the tenant it receives from hidden runtime context.""" + + def factory(seen_tenants: list[Any], *, approval_mode: Literal["always_require", "never_require"]) -> Agent[Any]: + @tool(name="read_tenant_data", approval_mode=approval_mode) + def read_tenant_data(ctx: FunctionInvocationContext) -> str: + seen_tenants.append(ctx.kwargs.get("tenant_id")) + return "ok" + + return Agent( + client=_ToolCallingClient(), + tools=[read_tenant_data], + default_options=cast(Any, {"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}), + ) + + return factory + + +@pytest.fixture +def tenant_workflow_agent_factory(tenant_agent_factory: TenantAgentFactory) -> TenantWorkflowAgentFactory: + """Wrap the tenant agent in a legacy ``WorkflowAgent`` so the workflow hosting path can be exercised.""" + + def factory(seen_tenants: list[Any]) -> WorkflowAgent: + @executor + async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None: + await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True)) + + inner = tenant_agent_factory(seen_tenants, approval_mode="never_require") + workflow = WorkflowBuilder(start_executor=start).add_edge(start, inner).build() + return WorkflowAgent(workflow=workflow, name="Tenant Workflow Agent") + + return factory diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py index d0b74669027..71e69a4a767 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py @@ -4,6 +4,7 @@ from __future__ import annotations +from collections.abc import Callable from typing import Any, Literal, cast from unittest.mock import AsyncMock, MagicMock @@ -11,13 +12,6 @@ import pytest from azure.ai.agentserver.responses import InMemoryResponseProvider, ResponseContext from azure.ai.agentserver.responses.models import CreateResponse -from tool_context_helpers import ( # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] - FORGED_TENANT_CONTEXT, - TRUSTED_TENANT_ID, - ToolCallingClient, - make_tenant_agent, - make_tenant_workflow_agent, -) from agent_framework_foundry_hosting import HostedResponseRequest, ResponsesHostServer from agent_framework_foundry_hosting._request import ( @@ -27,6 +21,9 @@ ) from agent_framework_foundry_hosting._scope import FoundryRequestScope +TRUSTED_TENANT_ID = "trusted" +FORGED_TENANT_CONTEXT: dict[str, Any] = {"additional_function_arguments": {"tenant_id": "forged"}} + @pytest.mark.parametrize( "payload", @@ -75,9 +72,9 @@ def _make_server(agent: Any, **kwargs: Any) -> ResponsesHostServer: @pytest.mark.xdist_group("session_isolation") class TestCallerCannotOverrideToolContext: - async def test_baseline_uses_trusted_tool_context(self) -> None: + async def test_baseline_uses_trusted_tool_context(self, tenant_agent_factory: Callable[..., Any]) -> None: seen: list[Any] = [] - server = _make_server(make_tenant_agent(seen, approval_mode="never_require")) + server = _make_server(tenant_agent_factory(seen, approval_mode="never_require")) response = await _post_json(server, {"input": "go", "stream": False}) @@ -87,10 +84,10 @@ async def test_baseline_uses_trusted_tool_context(self) -> None: @pytest.mark.parametrize("stream", [False, True]) @pytest.mark.parametrize("location", ["top_level", "extra_body"]) async def test_forged_tool_context_is_rejected_before_any_tool_runs( - self, location: Literal["top_level", "extra_body"], stream: bool + self, tenant_agent_factory: Callable[..., Any], location: Literal["top_level", "extra_body"], stream: bool ) -> None: seen: list[Any] = [] - agent = make_tenant_agent(seen, approval_mode="never_require") + agent = tenant_agent_factory(seen, approval_mode="never_require") server = _make_server(agent) payload: dict[str, Any] = {"input": "go", "stream": stream} if location == "top_level": @@ -103,15 +100,15 @@ async def test_forged_tool_context_is_rejected_before_any_tool_runs( assert "additional_function_arguments" in response.text assert "failed" in response.text assert seen == [] - assert cast(ToolCallingClient, agent.client).calls == 0 + assert cast(Any, agent.client).calls == 0 @pytest.mark.filterwarnings("ignore:Hosting WorkflowAgent through agent=.*:DeprecationWarning") @pytest.mark.parametrize("location", ["top_level", "extra_body"]) async def test_forged_tool_context_is_rejected_for_legacy_workflow_agents( - self, location: Literal["top_level", "extra_body"] + self, tenant_workflow_agent_factory: Callable[..., Any], location: Literal["top_level", "extra_body"] ) -> None: seen: list[Any] = [] - server = _make_server(make_tenant_workflow_agent(seen)) + server = _make_server(tenant_workflow_agent_factory(seen)) payload: dict[str, Any] = {"input": "go", "stream": False} if location == "top_level": payload.update(FORGED_TENANT_CONTEXT) @@ -125,19 +122,21 @@ async def test_forged_tool_context_is_rejected_for_legacy_workflow_agents( assert seen == [] @pytest.mark.filterwarnings("ignore:Hosting WorkflowAgent through agent=.*:DeprecationWarning") - async def test_legacy_workflow_agent_without_forged_context_still_runs(self) -> None: + async def test_legacy_workflow_agent_without_forged_context_still_runs( + self, tenant_workflow_agent_factory: Callable[..., Any] + ) -> None: seen: list[Any] = [] - server = _make_server(make_tenant_workflow_agent(seen)) + server = _make_server(tenant_workflow_agent_factory(seen)) response = await _post_json(server, {"input": "go", "stream": False}) assert response.json()["status"] == "completed" assert seen == [TRUSTED_TENANT_ID] - async def test_hook_can_still_supply_trusted_tool_context(self) -> None: + async def test_hook_can_still_supply_trusted_tool_context(self, tenant_agent_factory: Callable[..., Any]) -> None: seen: list[Any] = [] server = _make_server( - make_tenant_agent(seen, approval_mode="never_require"), + tenant_agent_factory(seen, approval_mode="never_require"), prepare_options=lambda request, options: { **options, "additional_function_arguments": {"tenant_id": f"from-hook:{request.scope.user_id}"}, @@ -150,9 +149,11 @@ async def test_hook_can_still_supply_trusted_tool_context(self) -> None: assert len(seen) == 1 assert str(seen[0]).startswith("from-hook:") - async def test_forged_context_on_approval_resume_is_rejected_without_consuming_approval(self) -> None: + async def test_forged_context_on_approval_resume_is_rejected_without_consuming_approval( + self, tenant_agent_factory: Callable[..., Any] + ) -> None: seen: list[Any] = [] - server = _make_server(make_tenant_agent(seen, approval_mode="always_require")) + server = _make_server(tenant_agent_factory(seen, approval_mode="always_require")) first = await _post_json(server, {"input": "go", "stream": False, "store": True}) first_body = first.json() diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py b/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py deleted file mode 100644 index f842ab5e0d8..00000000000 --- a/python/packages/foundry_hosting/tests/foundry_hosting/tool_context_helpers.py +++ /dev/null @@ -1,97 +0,0 @@ -# Copyright (c) Microsoft. All rights reserved. - -"""Deterministic agent helpers for hosted tool-execution-context tests.""" - -from __future__ import annotations - -from collections.abc import AsyncIterator, Awaitable, Mapping, Sequence -from typing import Any, Literal, cast - -from agent_framework import ( - Agent, - AgentExecutorRequest, - BaseChatClient, - ChatMiddlewareLayer, - ChatResponse, - ChatResponseUpdate, - Content, - FunctionInvocationContext, - FunctionInvocationLayer, - Message, - ResponseStream, - WorkflowAgent, - WorkflowBuilder, - WorkflowContext, - executor, - tool, -) - -TRUSTED_TENANT_ID = "trusted" -FORGED_TENANT_CONTEXT: dict[str, Any] = {"additional_function_arguments": {"tenant_id": "forged"}} - - -class ToolCallingClient( - FunctionInvocationLayer[Any], - ChatMiddlewareLayer[Any], - BaseChatClient[Any], -): - """Call ``read_tenant_data`` once, then answer.""" - - def __init__(self) -> None: - super().__init__(middleware=[]) - self.calls = 0 - - def _inner_get_response( - self, - *, - messages: Sequence[Message], - stream: bool, - options: Mapping[str, Any], - **kwargs: Any, - ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: - del options, kwargs - assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." - self.calls += 1 - needs_tool_call = not any( - content.type == "function_result" for message in messages for content in message.contents - ) - - async def stream_response() -> AsyncIterator[ChatResponseUpdate]: - if needs_tool_call: - yield ChatResponseUpdate( - contents=[Content.from_function_call(call_id="call_1", name="read_tenant_data", arguments="{}")], - role="assistant", - ) - else: - yield ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant") - - return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) - - -def make_tenant_agent( - seen_tenants: list[Any], *, approval_mode: Literal["always_require", "never_require"] -) -> Agent[Any]: - """Create an agent whose tool records the tenant it receives from hidden runtime context.""" - - @tool(name="read_tenant_data", approval_mode=approval_mode) - def read_tenant_data(ctx: FunctionInvocationContext) -> str: - seen_tenants.append(ctx.kwargs.get("tenant_id")) - return "ok" - - return Agent( - client=ToolCallingClient(), - tools=[read_tenant_data], - default_options=cast(Any, {"additional_function_arguments": {"tenant_id": TRUSTED_TENANT_ID}}), - ) - - -def make_tenant_workflow_agent(seen_tenants: list[Any]) -> WorkflowAgent: - """Wrap the tenant agent in a legacy ``WorkflowAgent`` so the workflow hosting path can be exercised.""" - - @executor - async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None: - await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True)) - - inner = make_tenant_agent(seen_tenants, approval_mode="never_require") - workflow = WorkflowBuilder(start_executor=start).add_edge(start, inner).build() - return WorkflowAgent(workflow=workflow, name="Tenant Workflow Agent") From 1bb4b33c2cde16f355e91df1beebeaaf1b377920 Mon Sep 17 00:00:00 2001 From: Darren Cohen <39422044+dargilco@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:39:22 -0700 Subject: [PATCH 7/7] Remove redundant cast flagged by ty in tool-context test Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../tests/foundry_hosting/test_caller_tool_context.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py index 71e69a4a767..bc4682d8500 100644 --- a/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py +++ b/python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py @@ -100,7 +100,7 @@ async def test_forged_tool_context_is_rejected_before_any_tool_runs( assert "additional_function_arguments" in response.text assert "failed" in response.text assert seen == [] - assert cast(Any, agent.client).calls == 0 + assert agent.client.calls == 0 @pytest.mark.filterwarnings("ignore:Hosting WorkflowAgent through agent=.*:DeprecationWarning") @pytest.mark.parametrize("location", ["top_level", "extra_body"])