diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 71dc8f59140..fa00ea0e99f 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -500,6 +500,15 @@ that manually replay messages own the equivalent rule: do not resend an approval approval `Message`, approval `Content`, or an earlier returned response. - Approval-time `UserInputRequiredException` and `MiddlewareTermination` return immediately without another model call. +- `Agent.as_tool()` keeps child function approvals inside the delegated invocation. A child + `ToolApprovalMiddleware` may resolve them through runtime `auto_approval_rules`; any unresolved child function + approval does not enter the caller's approval state or model transcript. An approval-only response fails the + agent-tool invocation; a mixed response preserves its non-approval user-input requests while discarding the child + approval continuation. Interactive, delayed, or durable approval belongs in a workflow. When + `propagate_session=True`, child application-state changes merge back into the parent while framework approval and + invocation-budget state remain isolated, including approval queues stored under custom child middleware + `source_id` values. Parent and child `ToolApprovalMiddleware` instances must use distinct `source_id` values; an + overlap fails before the child runs. ### Approval control content @@ -578,6 +587,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Mixed approved/rejected batch | Every call gets one correctly correlated terminal result. | `packages/core/tests/core/test_function_invocation_logic.py::test_rejected_approval` | | Persisted approval replay | Resume executes with the prior call available. | `test_persisted_approval_messages_replay_correctly` | | Hosted approval pass-through | Hosted requests/responses are bound to the recorded provider request and are not processed as local calls. | `test_hosted_tool_approval_response`, `test_hosted_mcp_approval_response_passthrough`, `test_session_approval_binding_reconstructs_hosted_response`, `test_mixed_local_and_hosted_approval_flow` | +| Agent-tool child approval | A child `ToolApprovalMiddleware` can auto-approve runtime tool requests inside one delegated invocation; unresolved child approvals execute nothing and do not enter the caller's approval state or dispatch a same-named parent tool, including with propagated application state. Mixed batches preserve non-approval user-input requests. Shared parent and child state rejects overlapping `ToolApprovalMiddleware.source_id` values before running the child, and custom child approval queues do not leak into later delegations. | `packages/core/tests/core/test_agents.py::test_chat_agent_as_tool_auto_approves_child_tool_with_middleware`, `test_chat_agent_as_tool_fails_closed_for_unresolved_child_approval`, `test_chat_agent_as_tool_child_approval_does_not_dispatch_same_named_parent_tool`, `test_chat_agent_as_tool_preserves_non_approval_requests_from_mixed_child_batch`, `test_chat_agent_as_tool_shared_session_requires_distinct_tool_approval_source_ids`, `test_chat_agent_as_tool_approved_delegation_does_not_confuse_framework_approval_state`, `test_chat_agent_as_tool_does_not_restore_custom_approval_queue_on_fresh_delegation` | | Approval-time user input | Every user-input request from one approved execution returns in order with assistant role and no extra model call; the execution consumes one call-budget unit. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_returns_all_user_input_requests_without_another_model_call`, `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_user_input_counts_toward_function_call_budget` | | Mixed terminal result and follow-up input | Completed siblings remain tool-role while only follow-up input requests use assistant-role messages/updates. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_separates_terminal_results_from_follow_up_requests`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_mixed_approval_resume_roles_serialize_function_result_as_tool`, `packages/core/tests/core/test_harness_tool_approval.py::test_dynamic_policy_approval_partitions_safe_sibling_result_roles` | | Approval-time middleware termination | Terminal result returns with no extra model call in either response mode. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_honors_middleware_termination` | diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index d9dcac20abf..9a4a1a8628a 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -63,7 +63,12 @@ map_chat_to_agent_update, normalize_messages, ) -from .exceptions import AgentInvalidRequestException, AgentInvalidResponseException, UserInputRequiredException +from .exceptions import ( + AgentInvalidRequestException, + AgentInvalidResponseException, + ToolExecutionException, + UserInputRequiredException, +) from .observability import AgentTelemetryLayer if sys.version_info >= (3, 13): @@ -95,6 +100,37 @@ # nested ``agent.run()`` (fresh options, its own session) keeps its own turn, # and nothing leaks into the caller's context while a stream is paused. _LOOP_ITERATION_TOKEN_KEY = "_agent_loop_iteration" # nosec B105 - a context-options key, not a credential # ruff: ignore[hardcoded-password-string] +_DELEGATED_STATE_MISSING = object() + + +def _tool_approval_source_ids(middleware: Sequence[MiddlewareTypes] | None) -> frozenset[str]: + """Return session-state keys owned by ToolApprovalMiddleware instances.""" + from ._harness._tool_approval import ToolApprovalMiddleware + + return frozenset(item.source_id for item in middleware or () if isinstance(item, ToolApprovalMiddleware)) + + +def _merge_delegated_session_state( + parent_state: MutableMapping[str, Any], + initial_child_state: Mapping[str, Any], + final_child_state: Mapping[str, Any], + *, + excluded_keys: frozenset[str], +) -> None: + """Merge child application-state changes without copying framework continuation state.""" + for key, initial_value in initial_child_state.items(): + if key in excluded_keys or key in final_child_state: + continue + if parent_state.get(key, _DELEGATED_STATE_MISSING) is initial_value: + parent_state.pop(key, None) + + for key, final_value in final_child_state.items(): + if key in excluded_keys: + continue + initial_value = initial_child_state.get(key, _DELEGATED_STATE_MISSING) + if initial_value is _DELEGATED_STATE_MISSING or final_value is not initial_value: + parent_state[key] = final_value + if TYPE_CHECKING: ResponseModelBoundT = TypeVar("ResponseModelBoundT", bound=BaseModel) @@ -624,12 +660,26 @@ def as_tool( approval_mode: Whether this delegated tool requires approval before execution. stream_callback: Optional callback for streaming responses. If provided, uses run(..., stream=True). propagate_session: If True, the parent agent's session is forwarded - to this sub-agent's ``run()`` call so both agents share the - same session. Defaults to False. + to this sub-agent's ``run()`` call. Application-state changes + propagate back to the parent, while framework approval continuation + state remains isolated. Defaults to False. The sub-agent always + receives an AgentSession so session-backed middleware can run. + When False, that session is private to this invocation. Returns: A FunctionTool that can be used as a tool by other agents. + Note: + Child function approvals are not propagated into the calling agent. + Configure ToolApprovalMiddleware with runtime auto-approval rules on + the child for immediate policy decisions. Use a workflow when approval + is interactive, delayed, or durable. + + When parent and child both use ToolApprovalMiddleware with + ``propagate_session=True``, configure distinct middleware ``source_id`` + values. The delegated call raises ToolExecutionException before running + the child when their shared session-state keys overlap. + Examples: .. code-block:: python @@ -676,39 +726,113 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str: ctx: the function invocation context used **kwargs: only used to dynamically load the argument that is defined for this tool. """ - session = ctx.session if propagate_session else None - - # Create a child session that shares the parent's state dict but has - # an isolated service_session_id. This avoids mutating the parent - # session in-place, which would race under concurrent asyncio.gather - # tool invocations sharing the same session. - if session is not None: - child_session = AgentSession(session_id=session.session_id) - child_session.state = session.state # shared by reference - child_session.service_session_id = None - session = child_session - - stream = self.run( - str(kwargs.get(arg_name, "")), - stream=True, - session=session, - function_invocation_kwargs=dict(ctx.kwargs), - ) - if stream_callback is not None: - # The callback is a host-facing observer: feed it the *released* - # updates by consuming the stream, never by registering a transform - # hook on it. Hooks can end up applied to buffered content ahead of an - # egress gate's verdict (see ResponseStream.buffered_and_gated), so a - # hook-registered observer could see denied or unredacted content. - async for update in stream: - callback_result = stream_callback(update) - if isawaitable(callback_result): - await callback_result - final_response = await stream.get_final_response() - if final_response.user_input_requests: - raise UserInputRequiredException(contents=final_response.user_input_requests) - # TODO(Copilot): update once #4331 merges - return final_response.text + parent_session = ctx.session + session = AgentSession() + child_approval_source_ids = _tool_approval_source_ids(self.middleware) + + if propagate_session and parent_session is not None: + from ._tools import _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY # pyright: ignore[reportPrivateUsage] + + raw_parent_approval_source_ids = ctx.metadata.get(_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY) + parent_approval_source_ids: frozenset[str] + parent_approval_source_ids = ( + cast("frozenset[str]", raw_parent_approval_source_ids) + if isinstance(raw_parent_approval_source_ids, frozenset) + else frozenset() + ) + overlapping_source_ids = child_approval_source_ids.intersection(parent_approval_source_ids) + if overlapping_source_ids: + formatted_source_ids = ", ".join(repr(source_id) for source_id in sorted(overlapping_source_ids)) + raise ToolExecutionException( + f"Agent tool {tool_name!r} cannot share its parent session because parent and child " + f"ToolApprovalMiddleware instances use the same source_id: {formatted_source_ids}. " + "Configure distinct source_id values or set propagate_session=False." + ) + + parent_state: MutableMapping[str, Any] | None = None + initial_child_state: dict[str, Any] | None = None + excluded_state_keys: frozenset[str] = frozenset() + + # Propagate application state through a child-owned copy. Framework + # approval continuation state stays isolated so an unresolved child + # request can never become pending authority in the parent session. + if propagate_session and parent_session is not None: + from ._tools import ( + _FUNCTION_INVOCATION_BUDGET_STATE_KEY, # pyright: ignore[reportPrivateUsage] + _FUNCTION_RESULT_PAYLOAD_BUDGET_STATE_KEY, # pyright: ignore[reportPrivateUsage] + _TOOL_APPROVAL_STATE_KEY, # pyright: ignore[reportPrivateUsage] + ) + + excluded_state_keys = frozenset({ + _TOOL_APPROVAL_STATE_KEY, + _FUNCTION_INVOCATION_BUDGET_STATE_KEY, + _FUNCTION_RESULT_PAYLOAD_BUDGET_STATE_KEY, + *child_approval_source_ids, + }) + parent_state = parent_session.state + child_state = {key: value for key, value in parent_state.items() if key not in excluded_state_keys} + initial_child_state = dict(child_state) + session = AgentSession(session_id=parent_session.session_id) + session.state = child_state + + try: + stream = self.run( + str(kwargs.get(arg_name, "")), + stream=True, + session=session, + function_invocation_kwargs=dict(ctx.kwargs), + ) + if stream_callback is not None: + # The callback is a host-facing observer: feed it the *released* + # updates by consuming the stream, never by registering a transform + # hook on it. Hooks can end up applied to buffered content ahead of an + # egress gate's verdict (see ResponseStream.buffered_and_gated), so a + # hook-registered observer could see denied or unredacted content. + async for update in stream: + callback_result = stream_callback(update) + if isawaitable(callback_result): + await callback_result + final_response = await stream.get_final_response() + approval_requests = [ + request + for request in final_response.user_input_requests + if request.type == "function_approval_request" + ] + other_input_requests = [ + request + for request in final_response.user_input_requests + if request.type != "function_approval_request" + ] + if approval_requests: + requested_tools = sorted( + { + request.function_call.name or "" + for request in approval_requests + if request.function_call is not None + } + or {""} + ) + approval_error = ( + f"Agent tool {tool_name!r} cannot continue because its sub-agent requested approval for " + f"{', '.join(requested_tools)}. Configure ToolApprovalMiddleware with auto_approval_rules on " + "the sub-agent for immediate policy decisions. Use a workflow for interactive, delayed, or " + "durable approval." + ) + if other_input_requests: + raise UserInputRequiredException(contents=other_input_requests, message=approval_error) + raise ToolExecutionException(approval_error) + if other_input_requests: + raise UserInputRequiredException(contents=other_input_requests) + # TODO(Copilot): update once #4331 merges + return final_response.text + finally: + if parent_state is not None and initial_child_state is not None: + _merge_delegated_session_state( + parent_state, + initial_child_state, + session.state, + excluded_keys=excluded_state_keys, + ) from ._tools import FunctionTool @@ -1450,6 +1574,7 @@ async def _prepare_run_context( agent_name = self._get_agent_name() from ._mcp import MCPTool + from ._tools import _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY # pyright: ignore[reportPrivateUsage] base_tools = _normalize_tools(chat_options.pop("tools", None)) mcp_duplicate_message = "Tool names must be unique. Consider setting `tool_name_prefix` on the MCPTool." @@ -1494,6 +1619,10 @@ async def _prepare_run_context( duplicate_error_message=mcp_duplicate_message, ) + additional_function_arguments[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = _tool_approval_source_ids( + self.middleware + ) + model = opts.pop("model", None) # Build options dict from run() options merged with provided options diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index ff2756cc60d..45f3d39b08b 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -106,6 +106,7 @@ def _generate_function_call_occurrence_id() -> str: SHELL_TOOL_KIND_VALUE: Final[str] = "shell" _TOOL_APPROVAL_STATE_KEY: Final[str] = "tool_approval" _APPROVAL_SESSION_IS_AUTHORITATIVE_KEY: Final[str] = "_approval_session_is_authoritative" +_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY: Final[str] = "_parent_tool_approval_source_ids" def _has_authoritative_approval_session(invocation_session: AgentSession | None) -> bool: @@ -2068,8 +2069,21 @@ async def _auto_invoke_function( runtime_kwargs: dict[str, Any] = { key: value for key, value in (custom_args or {}).items() - if key not in {"_function_middleware_pipeline", "middleware", "conversation_id"} + if key + not in { + "_function_middleware_pipeline", + "middleware", + "conversation_id", + _PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY, + } } + raw_parent_approval_source_ids = (custom_args or {}).get(_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY) + parent_approval_source_ids: frozenset[str] + parent_approval_source_ids = ( + cast("frozenset[str]", raw_parent_approval_source_ids) + if isinstance(raw_parent_approval_source_ids, frozenset) + else frozenset() + ) if invocation_session is not None: runtime_kwargs["session"] = invocation_session args = dict(parsed_args) @@ -2088,6 +2102,7 @@ async def _auto_invoke_function( kwargs=runtime_kwargs.copy(), tools=live_tools, ) + direct_context.metadata[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = parent_approval_source_ids if host_payload_budget is not None: direct_context.metadata[_FUNCTION_RESULT_PAYLOAD_BUDGET_CONTEXT_KEY] = host_payload_budget function_result = await tool.invoke( @@ -2126,6 +2141,7 @@ async def _auto_invoke_function( kwargs=runtime_kwargs.copy(), tools=live_tools, ) + middleware_context.metadata[_PARENT_TOOL_APPROVAL_SOURCE_IDS_CONTEXT_KEY] = parent_approval_source_ids if host_payload_budget is not None: middleware_context.metadata[_FUNCTION_RESULT_PAYLOAD_BUDGET_CONTEXT_KEY] = host_payload_budget middleware_context.metadata[_AUTO_ARGUMENT_PREPARATION_CONTEXT_KEY] = True @@ -3061,9 +3077,7 @@ def _stage_approval_batch_responses( if any(request_id not in stored_responses for request_id in group_ids): updated_group = dict(group) updated_group[_APPROVAL_RESPONSES_KEY] = [ - stored_responses[request_id].to_dict() - for request_id in group_ids - if request_id in stored_responses + stored_responses[request_id].to_dict() for request_id in group_ids if request_id in stored_responses ] remaining_groups.append(updated_group) missing_request_ids = [request_id for request_id in group_ids if request_id not in stored_responses] diff --git a/python/packages/core/agent_framework/exceptions.py b/python/packages/core/agent_framework/exceptions.py index 03f950e331a..a908b5bce5d 100644 --- a/python/packages/core/agent_framework/exceptions.py +++ b/python/packages/core/agent_framework/exceptions.py @@ -188,12 +188,12 @@ class ToolExecutionException(ToolException): class UserInputRequiredException(ToolException): - """Raised when a tool wrapping a sub-agent requires user input to proceed. + """Raised when a tool requires user input to proceed. - This exception carries the ``user_input_request`` Content items emitted by - the sub-agent (e.g., ``oauth_consent_request``, ``function_approval_request``) - so the tool invocation layer can propagate them to the parent agent's response - instead of swallowing them as a generic tool error. + This exception carries ``user_input_request`` Content items so the tool + invocation layer can propagate them instead of swallowing them as a generic + tool error. ``Agent.as_tool()`` handles unresolved child function approvals + separately and fails the delegated invocation. Args: contents: The user-input-request Content items from the sub-agent response. diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index 612e759b9af..e54489c624f 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -41,6 +41,7 @@ SlidingWindowStrategy, SupportsAgentRun, SupportsChatGetResponse, + ToolApprovalMiddleware, ToolResultCompactionStrategy, TruncationStrategy, VectorStoreHistoryProvider, @@ -51,7 +52,11 @@ from agent_framework._agents import _get_tool_name, _merge_options, _sanitize_agent_name from agent_framework._mcp import MCPTool, _build_prefixed_mcp_name, _normalize_mcp_name from agent_framework._middleware import FunctionInvocationContext -from agent_framework.exceptions import AgentInvalidRequestException, ChatClientInvalidResponseException +from agent_framework.exceptions import ( + AgentInvalidRequestException, + ChatClientInvalidResponseException, + ToolExecutionException, +) from .conftest import MockBaseChatClient @@ -1698,6 +1703,296 @@ async def test_chat_agent_as_tool_function_execution( assert result[0].text == "test streaming response another update" # From mock streaming client +async def test_chat_agent_as_tool_auto_approves_child_tool_with_middleware() -> None: + """Test that child ToolApprovalMiddleware policies resolve approval during the delegated invocation.""" + executions: list[str] = [] + observed_calls: list[Content] = [] + + @tool(name="read_weather", approval_mode="always_require") + def read_weather(location: str) -> str: + executions.append(location) + return f"Weather for {location}" + + def approve_weather(function_call: Content) -> bool: + observed_calls.append(function_call) + return function_call.name == "read_weather" and function_call.parse_arguments() == {"location": "Amsterdam"} + + client = MockBaseChatClient() + client.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call( + call_id="weather-call", + name="read_weather", + arguments={"location": "Amsterdam"}, + ) + ], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("The weather is cloudy.")])], + ] + agent = Agent( + client=client, + name="WeatherAgent", + tools=[read_weather], + middleware=[ToolApprovalMiddleware(auto_approval_rules=[approve_weather])], + ) + + result = await agent.as_tool().invoke(arguments={"task": "Check Amsterdam weather"}) + + assert executions == ["Amsterdam"] + assert len(observed_calls) == 1 + assert observed_calls[0].parse_arguments() == {"location": "Amsterdam"} + assert result[0].text == "The weather is cloudy." + + +async def test_chat_agent_as_tool_fails_closed_for_unresolved_child_approval() -> None: + """Test that an unresolved child approval cannot escape into the caller's tool registry.""" + executions = 0 + + @tool(name="delete_weather_data", approval_mode="always_require") + def delete_weather_data() -> str: + nonlocal executions + executions += 1 + return "deleted" + + client = MockBaseChatClient() + client.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call( + call_id="delete-call", + name="delete_weather_data", + arguments={}, + ) + ], + ) + ] + ] + agent = Agent( + client=client, + name="WeatherAgent", + tools=[delete_weather_data], + middleware=[ToolApprovalMiddleware(auto_approval_rules=[lambda function_call: False])], + ) + + with raises( + ToolExecutionException, + match=( + "sub-agent requested approval for delete_weather_data.*" + "ToolApprovalMiddleware.*Use a workflow for interactive, delayed, or durable approval" + ), + ): + await agent.as_tool().invoke(arguments={"task": "Delete weather data"}) + + assert executions == 0 + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("propagate_session", [False, True]) +async def test_chat_agent_as_tool_child_approval_does_not_dispatch_same_named_parent_tool( + stream: bool, + propagate_session: bool, +) -> None: + """Test that child approval cannot escape and bind to a same-named parent tool.""" + child_executions = 0 + parent_executions = 0 + observed_approval_requests: list[Content] = [] + + @tool(name="guarded_write", approval_mode="always_require") + def child_guarded_write() -> str: + nonlocal child_executions + child_executions += 1 + return "child" + + @tool(name="guarded_write", approval_mode="never_require") + def parent_guarded_write() -> str: + nonlocal parent_executions + parent_executions += 1 + return "parent" + + child_client = MockBaseChatClient() + child_client.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call( + call_id="child-write", + name="guarded_write", + arguments={}, + ) + ], + ) + ] + ] + child_agent = Agent(client=child_client, name="ChildAgent", tools=[child_guarded_write]) + + parent_client = MockBaseChatClient() + parent_call = Content.from_function_call( + call_id="delegate-call", + name="delegate", + arguments={"task": "write"}, + ) + + def observe_child_updates(update: AgentResponseUpdate) -> None: + observed_approval_requests.extend(update.user_input_requests) + + if stream: + parent_client.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[parent_call])], + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_text("The delegated write was blocked.")], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("The stale approval was ignored.")])], + ] + else: + parent_client.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[parent_call], + ) + ), + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_text("The delegated write was blocked.")], + ) + ), + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_text("The stale approval was ignored.")], + ) + ), + ] + parent_agent = Agent( + client=parent_client, + name="ParentAgent", + tools=[ + child_agent.as_tool( + name="delegate", + propagate_session=propagate_session, + stream_callback=observe_child_updates, + ), + parent_guarded_write, + ], + ) + parent_session = AgentSession() + + if stream: + result = await parent_agent.run( + "Delegate the write.", + session=parent_session, + stream=True, + ).get_final_response() + else: + result = await parent_agent.run( + "Delegate the write.", + session=parent_session, + stream=False, + ) + + assert child_executions == 0 + assert parent_executions == 0 + assert not result.user_input_requests + assert result.text == "The delegated write was blocked." + + if propagate_session: + assert len(observed_approval_requests) == 1 + tool_approval_state = parent_session.state.get("tool_approval", {}) + assert isinstance(tool_approval_state, dict) + assert not tool_approval_state.get("pending_approval_requests") + + stale_approval = observed_approval_requests[0].to_function_approval_response(approved=True) + if stream: + stale_result = await parent_agent.run( + stale_approval, + session=parent_session, + stream=True, + ).get_final_response() + else: + stale_result = await parent_agent.run( + stale_approval, + session=parent_session, + stream=False, + ) + + assert stale_result.text == "The stale approval was ignored." + assert child_executions == 0 + assert parent_executions == 0 + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_chat_agent_as_tool_preserves_non_approval_requests_from_mixed_child_batch(stream: bool) -> None: + """Test that fail-closed child approvals do not discard other user-input requests.""" + + @tool(name="guarded_write", approval_mode="always_require") + def guarded_write() -> str: + raise AssertionError("Unapproved child tool must not execute.") + + child_client = MockBaseChatClient() + child_client.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call(call_id="child-write", name="guarded_write", arguments={}), + Content.from_oauth_consent_request(consent_link="https://example.com/consent"), + ], + ) + ] + ] + child_agent = Agent(client=child_client, name="ChildAgent", tools=[guarded_write]) + + parent_client = MockBaseChatClient() + parent_call = Content.from_function_call( + call_id="delegate-call", + name="delegate", + arguments={"task": "write"}, + ) + if stream: + parent_client.streaming_responses = [[ChatResponseUpdate(role="assistant", contents=[parent_call])]] + else: + parent_client.run_responses = [ + ChatResponse(messages=Message(role="assistant", contents=[parent_call])), + ] + parent_agent = Agent( + client=parent_client, + name="ParentAgent", + tools=[child_agent.as_tool(name="delegate", propagate_session=True)], + ) + parent_session = AgentSession() + + if stream: + result = await parent_agent.run( + "Delegate the write.", + session=parent_session, + stream=True, + ).get_final_response() + else: + result = await parent_agent.run( + "Delegate the write.", + session=parent_session, + stream=False, + ) + + assert len(result.user_input_requests) == 1 + assert result.user_input_requests[0].type == "oauth_consent_request" + assert result.user_input_requests[0].consent_link == "https://example.com/consent" + tool_approval_state = parent_session.state.get("tool_approval", {}) + assert isinstance(tool_approval_state, dict) + assert not tool_approval_state.get("pending_approval_requests") + + async def test_chat_agent_as_tool_with_stream_callback( client: SupportsChatGetResponse, ) -> None: @@ -1810,18 +2105,215 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: ) ) - # Child receives a separate AgentSession (not the parent object) to isolate - # service_session_id, but shares the same state dict and session_id. + # Child receives a separate AgentSession and state mapping so framework + # continuation state stays isolated while application state propagates. assert captured_session is not None assert captured_session is not parent_session assert captured_session.session_id == "parent-session-123" - assert captured_session.state is parent_session.state + assert captured_session.state is not parent_session.state assert captured_session.state["shared_key"] == "shared_value" + assert parent_session.state["shared_key"] == "shared_value" assert captured_session.service_session_id is None -async def test_chat_agent_as_tool_propagate_session_false_by_default(client: SupportsChatGetResponse) -> None: - """Test that propagate_session defaults to False and does not forward the session.""" +@pytest.mark.parametrize( + ("child_source_id", "child_should_run"), + [ + ("tool_approval", False), + ("child_tool_approval", True), + ], +) +async def test_chat_agent_as_tool_shared_session_requires_distinct_tool_approval_source_ids( + child_source_id: str, + child_should_run: bool, +) -> None: + """Test that shared parent and child approval state cannot use the same key.""" + child_client = MockBaseChatClient() + child_agent = Agent( + client=child_client, + name="ChildAgent", + middleware=[ToolApprovalMiddleware(source_id=child_source_id)], + ) + child_run_called = False + original_child_run = child_agent.run + + def capturing_child_run(*args: Any, **kwargs: Any) -> Any: + nonlocal child_run_called + child_run_called = True + return original_child_run(*args, **kwargs) + + child_agent.run = capturing_child_run # type: ignore[assignment, method-assign] # ty: ignore[invalid-assignment] + + parent_client = MockBaseChatClient() + parent_client.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="delegate-call", + name="delegate", + arguments={"task": "Complete the child task"}, + ) + ], + ) + ), + ChatResponse(messages=Message(role="assistant", contents=[Content.from_text("Done.")])), + ] + parent_agent = Agent( + client=parent_client, + name="ParentAgent", + middleware=[ToolApprovalMiddleware()], + tools=[child_agent.as_tool(name="delegate", propagate_session=True)], + ) + + result = await parent_agent.run("Delegate the task.", session=AgentSession()) + + assert child_run_called is child_should_run + assert result.text == "Done." + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_chat_agent_as_tool_approved_delegation_does_not_confuse_framework_approval_state( + stream: bool, +) -> None: + """Test that ordinary parent approval state is not treated as middleware ownership.""" + child_client = MockBaseChatClient() + child_client.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Child completed.")])] + ] + child_agent = Agent( + client=child_client, + name="ChildAgent", + middleware=[ToolApprovalMiddleware()], + ) + + parent_client = MockBaseChatClient() + parent_call = Content.from_function_call( + call_id="delegate-call", + name="delegate", + arguments={"task": "Complete the child task"}, + ) + if stream: + parent_client.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[parent_call])], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Parent completed.")])], + ] + else: + parent_client.run_responses = [ + ChatResponse(messages=Message(role="assistant", contents=[parent_call])), + ChatResponse(messages=Message(role="assistant", contents=[Content.from_text("Parent completed.")])), + ] + parent_agent = Agent( + client=parent_client, + name="ParentAgent", + tools=[ + child_agent.as_tool( + name="delegate", + approval_mode="always_require", + propagate_session=True, + ) + ], + ) + parent_session = AgentSession() + + if stream: + first_response = await parent_agent.run( + "Delegate the task.", + session=parent_session, + stream=True, + ).get_final_response() + else: + first_response = await parent_agent.run( + "Delegate the task.", + session=parent_session, + stream=False, + ) + + approval_response = first_response.user_input_requests[0].to_function_approval_response(approved=True) + if stream: + result = await parent_agent.run( + approval_response, + session=parent_session, + stream=True, + ).get_final_response() + else: + result = await parent_agent.run( + approval_response, + session=parent_session, + stream=False, + ) + + delegated_result = next( + content + for message in result.messages + for content in message.contents + if content.type == "function_result" and content.call_id == "delegate-call" + ) + assert delegated_result.result == "Child completed." + assert delegated_result.exception is None + assert child_client.call_count == 1 + assert result.text == "Parent completed." + + +async def test_chat_agent_as_tool_does_not_restore_custom_approval_queue_on_fresh_delegation() -> None: + """Test that custom child approval state cannot leak through a shared parent session.""" + + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + raise AssertionError("Unapproved child tool must not execute.") + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + raise AssertionError("Unapproved child tool must not execute.") + + child_client = MockBaseChatClient() + child_client.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call(call_id="first-write", name="first_write", arguments={}), + Content.from_function_call(call_id="second-write", name="second_write", arguments={}), + ], + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("Fresh delegation completed.")])], + ] + child_agent = Agent( + client=child_client, + name="ChildAgent", + tools=[first_write, second_write], + middleware=[ToolApprovalMiddleware(source_id="child_approval")], + ) + delegated_tool = child_agent.as_tool(propagate_session=True) + parent_session = AgentSession() + + with raises(ToolExecutionException, match="sub-agent requested approval"): + await delegated_tool.invoke( + context=FunctionInvocationContext( + function=delegated_tool, + arguments={"task": "First delegation"}, + session=parent_session, + ) + ) + + assert "child_approval" not in parent_session.state + + result = await delegated_tool.invoke( + context=FunctionInvocationContext( + function=delegated_tool, + arguments={"task": "Fresh delegation"}, + session=parent_session, + ) + ) + + assert result[0].text == "Fresh delegation completed." + assert not child_client.streaming_responses + + +async def test_chat_agent_as_tool_uses_private_session_by_default(client: SupportsChatGetResponse) -> None: + """Test that the default private session supports child middleware without sharing parent state.""" agent = Agent(client=client, name="SubAgent", description="Sub agent") tool = agent.as_tool() # default: propagate_session=False @@ -1845,7 +2337,10 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: ) ) - assert captured_session is None + assert captured_session is not None + assert captured_session is not parent_session + assert captured_session.state is not parent_session.state + assert captured_session.service_session_id is None async def test_chat_agent_as_tool_propagate_session_shares_state(client: SupportsChatGetResponse) -> None: @@ -1898,8 +2393,8 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: assert captured_session is not None assert captured_session is not parent_session assert captured_session.service_session_id is None - # But shares the same state dict by reference - assert captured_session.state is parent_session.state + # Application state is copied into the child and merged back afterward. + assert captured_session.state is not parent_session.state assert captured_session.state["data"] == "shared" return original_run(*args, **kwargs) diff --git a/python/samples/03-workflows/README.md b/python/samples/03-workflows/README.md index b3cb399dd60..de0b7f73597 100644 --- a/python/samples/03-workflows/README.md +++ b/python/samples/03-workflows/README.md @@ -104,7 +104,11 @@ Builder-oriented request-info samples are maintained in the orchestration sample ### tool-approval -Builder-based tool approval samples are maintained in the orchestration sample set. +| Sample | File | Concepts | +| ------ | ---- | -------- | +| Agent Tool vs Workflow Approval | [tool-approval/agent_as_tool_vs_workflow_approval.py](./tool-approval/agent_as_tool_vs_workflow_approval.py) | Use child `ToolApprovalMiddleware` policy for immediate delegated approval; use a workflow for delayed or durable approval | + +Additional builder-based tool approval samples are maintained in the orchestration sample set. ### observability diff --git a/python/samples/03-workflows/tool-approval/agent_as_tool_vs_workflow_approval.py b/python/samples/03-workflows/tool-approval/agent_as_tool_vs_workflow_approval.py new file mode 100644 index 00000000000..c0e64199b76 --- /dev/null +++ b/python/samples/03-workflows/tool-approval/agent_as_tool_vs_workflow_approval.py @@ -0,0 +1,194 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Compare immediate agent-tool approval policy with durable workflow approval. + +Use ``Agent.as_tool()`` when the child can decide approvals immediately through +``ToolApprovalMiddleware.auto_approval_rules``. Use a workflow when a person or +external system may respond later. The workflow models delegation as a call and +return: the coordinator sends one task to the child, the child may pause for +approval, and the result returns to the same coordinator. + +Prerequisites: +- FOUNDRY_PROJECT_ENDPOINT: Microsoft Foundry project endpoint. +- FOUNDRY_MODEL: Model deployment name. +- Run ``az login`` before starting the sample. +""" + +import asyncio +import os +from collections.abc import AsyncIterable +from typing import Any, Literal + +from agent_framework import ( + Agent, + AgentExecutor, + AgentExecutorResponse, + Content, + ToolApprovalMiddleware, + WorkflowBuilder, + WorkflowContext, + WorkflowEvent, + executor, + tool, +) +from agent_framework.foundry import FoundryChatClient +from agent_framework.openai import OpenAIChatOptions +from azure.identity import AzureCliCredential +from pydantic import BaseModel +from typing_extensions import Never + + +@tool(approval_mode="always_require") +def reserve_inventory(item: str, quantity: int) -> str: + """Reserve a quantity of an inventory item.""" + return f"Reserved {quantity} unit(s) of {item}." + + +def approve_small_reservations(function_call: Content) -> bool: + """Approve only small inventory reservations during the current invocation. + + The middleware evaluates this policy when a request appears, so the same + approach applies to tools discovered at runtime through MCP or Skills. + """ + if function_call.name != "reserve_inventory": + return False + arguments = function_call.parse_arguments() or {} + quantity = arguments.get("quantity") + return isinstance(quantity, int) and 0 < quantity <= 5 + + +def create_client() -> FoundryChatClient: + """Create the chat client used by both examples.""" + return FoundryChatClient( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + model=os.environ["FOUNDRY_MODEL"], + credential=AzureCliCredential(), + ) + + +async def run_agent_tool_example() -> None: + """Run an immediate approval policy entirely inside Agent.as_tool().""" + inventory_agent = Agent( + client=create_client(), + name="InventoryAgent", + instructions="Reserve the requested inventory using the available tool.", + tools=[reserve_inventory], + middleware=[ + ToolApprovalMiddleware( + auto_approval_rules=[approve_small_reservations], + ) + ], + ) + coordinator = Agent( + client=create_client(), + name="Coordinator", + instructions="Delegate inventory reservations to InventoryAgent.", + tools=[inventory_agent.as_tool()], + ) + + result = await coordinator.run("Reserve 2 keyboards.") + print("Agent.as_tool result:") + print(result.text) + + +async def collect_workflow_events(stream: AsyncIterable[WorkflowEvent]) -> dict[str, Content]: + """Collect approval requests and display completed workflow output.""" + requests: dict[str, Content] = {} + async for event in stream: + if event.type == "request_info" and isinstance(event.data, Content): + requests[event.request_id] = event.data + elif event.type == "output" and isinstance(event.data, str): + print("Workflow result:") + print(event.data) + return requests + + +class CoordinatorDecision(BaseModel): + """Choose whether to delegate a task or complete the caller's response.""" + + action: Literal["delegate", "complete"] + message: str + + +def chose_action(expected: Literal["delegate", "complete"]): + """Create an edge condition for a structured coordinator decision.""" + + def condition(response: Any) -> bool: + return ( + isinstance(response, AgentExecutorResponse) + and isinstance(response.agent_response.value, CoordinatorDecision) + and response.agent_response.value.action == expected + ) + + return condition + + +@executor(id="complete_reservation") +async def complete_reservation( + response: AgentExecutorResponse, + ctx: WorkflowContext[Never, str], +) -> None: + """Return the coordinator's completed response.""" + decision = response.agent_response.value + if not isinstance(decision, CoordinatorDecision): + raise ValueError("Coordinator response must be a CoordinatorDecision.") + await ctx.yield_output(decision.message) + + +async def run_workflow_example() -> None: + """Delegate to a child and return its result through a durable workflow.""" + coordinator = AgentExecutor( + Agent( + client=create_client(), + name="Coordinator", + instructions=( + "You coordinate inventory reservations. For a new request, set action to 'delegate' and message " + "to a concise task for InventoryAgent. When InventoryAgent returns a result, set action to " + "'complete' and message to a user-facing summary of that result." + ), + default_options=OpenAIChatOptions[Any](response_format=CoordinatorDecision), + ) + ) + inventory_agent = Agent( + client=create_client(), + name="InventoryAgent", + instructions="Reserve the requested inventory using the available tool.", + tools=[reserve_inventory], + ) + inventory = AgentExecutor(inventory_agent) + workflow = ( + WorkflowBuilder(start_executor=coordinator) + .add_edge(coordinator, inventory, condition=chose_action("delegate")) + .add_edge(inventory, coordinator) + .add_edge(coordinator, complete_reservation, condition=chose_action("complete")) + .build() + ) + + requests = await collect_workflow_events(workflow.run("Reserve 20 keyboards.", stream=True)) + while requests: + responses: dict[str, Content] = {} + for request_id, request in requests.items(): + if request.type != "function_approval_request" or request.function_call is None: + continue + print("Workflow paused for approval:") + print(f" Tool: {request.function_call.name}") + print(f" Arguments: {request.function_call.arguments}") + + # A real application can persist a workflow checkpoint and return much later. + await asyncio.sleep(1) + responses[request_id] = request.to_function_approval_response(approved=True) + + requests = await collect_workflow_events(workflow.run(stream=True, responses=responses)) + + +async def main() -> None: + """Run the immediate and delayed approval approaches in order.""" + print("1. Immediate policy with Agent.as_tool") + await run_agent_tool_example() + + print("\n2. Delayed approval with a workflow") + await run_workflow_example() + + +if __name__ == "__main__": + asyncio.run(main())