Python: [BREAKING] Reject caller-supplied tool execution context in Responses requests

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

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,10 +116,42 @@ 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 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."""
+    """Translate native generation fields, with flattened extra-body values winning collisions.
+
+    Raises:
+        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] = {}

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:

python/packages/foundry_hosting/tests/foundry_hosting/conftest.py

@@ -0,0 +1,106 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Fixtures for hosted tool-execution-context tests."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
+from typing import Any, Literal, cast
+
+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

python/packages/foundry_hosting/tests/foundry_hosting/test_caller_tool_context.py

@@ -0,0 +1,178 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Callers must not supply or replace trusted tool-execution context on Responses requests."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+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 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
+
+TRUSTED_TENANT_ID = "trusted"
+FORGED_TENANT_CONTEXT: dict[str, Any] = {"additional_function_arguments": {"tenant_id": "forged"}}
+
+
+@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, tenant_agent_factory: Callable[..., Any]) -> None:
+        seen: list[Any] = []
+        server = _make_server(tenant_agent_factory(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, tenant_agent_factory: Callable[..., Any], location: Literal["top_level", "extra_body"], stream: bool
+    ) -> None:
+        seen: list[Any] = []
+        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":
+            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 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, tenant_workflow_agent_factory: Callable[..., Any], location: Literal["top_level", "extra_body"]
+    ) -> None:
+        seen: list[Any] = []
+        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)
+        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, tenant_workflow_agent_factory: Callable[..., Any]
+    ) -> None:
+        seen: list[Any] = []
+        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, tenant_agent_factory: Callable[..., Any]) -> None:
+        seen: list[Any] = []
+        server = _make_server(
+            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}"},
+            },
+        )
+
+        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, tenant_agent_factory: Callable[..., Any]
+    ) -> None:
+        seen: list[Any] = []
+        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()
+        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]