diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index b1d8511e7a..e2573d189c 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -485,38 +485,6 @@ class WorkflowAgent(BaseAgent): # endregion Run Methods - def _process_request_info_event( - self, - event: WorkflowEvent[Any], - ) -> Content: - """Convert a request_info event to FunctionApprovalRequestContent. - - Args: - event: A WorkflowEvent with type='request_info'. - - Returns: - A FunctionApprovalRequestContent. - - Note: - If the event data is already a FunctionApprovalRequestContent, it will be returned as-is. - """ - if isinstance(event.data, Content) and event.data.user_input_request: - # Return the event data as-is if it's already a properly formed FunctionApprovalRequestContent - return event.data - - request_id = event.request_id - args = self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).to_dict() - - function_call = Content.from_function_call( - call_id=request_id, - name=self.REQUEST_INFO_FUNCTION_NAME, - arguments=args, - ) - return Content.from_function_approval_request( - id=request_id, - function_call=function_call, - ) - def _convert_workflow_events_to_agent_response( self, response_id: str, @@ -534,10 +502,10 @@ class WorkflowAgent(BaseAgent): for output_event in output_events: if output_event.type == "request_info": - approval_request = self._process_request_info_event(output_event) + request_content = self._process_request_info_event(output_event) messages.append( Message( - contents=[approval_request], + contents=[request_content], role="assistant", author_name=output_event.source_executor_id, message_id=str(uuid.uuid4()), @@ -705,10 +673,10 @@ class WorkflowAgent(BaseAgent): ] if event.type == "request_info": - approval_request = self._process_request_info_event(event) + request_content = self._process_request_info_event(event) return [ AgentResponseUpdate( - contents=[approval_request], + contents=[request_content], role="assistant", author_name=self.name, response_id=response_id, @@ -721,36 +689,48 @@ class WorkflowAgent(BaseAgent): # Ignore workflow-internal events return [] + def _process_request_info_event( + self, + event: WorkflowEvent[Any], + ) -> Content: + """Convert a request_info event to FunctionApprovalRequestContent. + + Args: + event: A WorkflowEvent with type='request_info'. + + Returns: + A content object representing the request info. The content can be a `function_approval_request` + or a `function_call` depending on the structure of the event data. + + Note: + If the event data is already a FunctionApprovalRequestContent, it will be returned as-is. + """ + if isinstance(event.data, Content) and event.data.user_input_request: + # Return the event data as-is if it's already a properly formed FunctionApprovalRequestContent + return event.data + + request_id = event.request_id + args = self.RequestInfoFunctionArgs(request_id=request_id, data=event).to_dict() + + return Content.from_function_call( + call_id=request_id, + name=self.REQUEST_INFO_FUNCTION_NAME, + arguments=args, + ) + def _extract_function_responses(self, input_messages: Sequence[Message]) -> dict[str, Any]: - """Extract function responses from input messages.""" + """Extract function responses from input messages. + + The responses are for pending requests that the workflow is waiting on, and + will be passed to the workflow. The pending requests are processed to either + `function_approval_request` or `function_call` content by `_process_request_info_event`. + """ function_responses: dict[str, Any] = {} for message in input_messages: for content in message.contents: if content.type == "function_approval_response": request_id: str = content.id # type: ignore[assignment] - function_call: Content = content.function_call # type: ignore[assignment] - # Parse the function arguments to recover request payload - if function_call.name != self.REQUEST_INFO_FUNCTION_NAME: - # This response is for a raw approval request that is itself already an - # approval request. - function_responses[request_id] = content - else: - # This response is for an approval request constructed from a request_info event. - arguments_payload = content.function_call.arguments # type: ignore[attr-defined, union-attr] - if isinstance(arguments_payload, str): - try: - parsed_args = self.RequestInfoFunctionArgs.from_json(arguments_payload) - except ValueError as exc: - raise AgentInvalidResponseException( - "FunctionApprovalResponseContent arguments must decode to a mapping." - ) from exc - elif isinstance(arguments_payload, dict): - parsed_args = self.RequestInfoFunctionArgs.from_dict(arguments_payload) - else: - raise AgentInvalidResponseException( - "FunctionApprovalResponseContent arguments must be a mapping or JSON string." - ) - function_responses[request_id] = parsed_args.data + function_responses[request_id] = content elif content.type == "function_result": response_data = content.result if hasattr(content, "result") else str(content) # type: ignore[attr-defined] function_responses[content.call_id] = response_data # type: ignore[argument-type] diff --git a/python/packages/core/tests/workflow/test_workflow_agent.py b/python/packages/core/tests/workflow/test_workflow_agent.py index 66ea6669fd..fe0c854f73 100644 --- a/python/packages/core/tests/workflow/test_workflow_agent.py +++ b/python/packages/core/tests/workflow/test_workflow_agent.py @@ -29,6 +29,7 @@ from agent_framework import ( handler, response_handler, ) +from agent_framework._workflows._typing_utils import deserialize_type class SimpleExecutor(Executor): @@ -239,50 +240,38 @@ class TestWorkflowAgent: # Should have received an approval request for the request info assert len(updates) > 0 - approval_update: AgentResponseUpdate | None = None + request_update: AgentResponseUpdate | None = None for update in updates: - if any(content.type == "function_approval_request" for content in update.contents): - approval_update = update + if any(content.type == "function_call" for content in update.contents): + request_update = update break - assert approval_update is not None, "Should have received a request_info approval request" + assert request_update is not None, "Should have received a request_info wrapped in a function_call content" - approval_request = next( - content for content in approval_update.contents if content.type == "function_approval_request" - ) - assert approval_request.id is not None - assert approval_request.function_call is not None - - function_call = approval_request.function_call + request_function_call = next(content for content in request_update.contents if content.type == "function_call") + assert request_function_call.call_id is not None # Verify the function call has expected structure - assert function_call.call_id is not None - assert function_call.name == "request_info" - assert isinstance(function_call.arguments, dict) - assert function_call.arguments.get("request_id") == approval_request.id + assert request_function_call.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME + assert isinstance(request_function_call.arguments, dict) + assert request_function_call.arguments.get("request_id") is not None + assert request_function_call.arguments.get("data") is not None + request_data = request_function_call.arguments["data"] + assert request_data.get("type") == "request_info" + assert deserialize_type(request_data.get("response_type")) is str # Verify the request is tracked in pending_requests pending_requests = await workflow._runner_context.get_pending_request_info_events() assert len(pending_requests) == 1 - assert function_call.call_id in pending_requests + assert request_function_call.call_id in pending_requests - # Now provide an approval response with updated arguments to test continuation - response_args = WorkflowAgent.RequestInfoFunctionArgs( - request_id=approval_request.id, - data="User provided answer", - ).to_dict() - - approval_response = Content.from_function_approval_response( - approved=True, - id=approval_request.id, - function_call=Content.from_function_call( - call_id=function_call.call_id, - name=function_call.name, - arguments=response_args, - ), + # Now provide a function result response with updated arguments to test continuation + function_result = Content.from_function_result( + call_id=request_function_call.call_id, + result="Mock response to request info", ) - response_message = Message(role="user", contents=[approval_response]) + response_message = Message(role="user", contents=[function_result]) # Continue the workflow with the response continuation_result = await agent.run(response_message) @@ -312,13 +301,359 @@ class TestWorkflowAgent: response_type=str, ) - approval_request = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage] + request_function_call = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage] - assert approval_request.function_call is not None - assert approval_request.function_call.arguments == { - "request_id": "request_123", - "data": {"target_agent": "helper", "reason": "overflow"}, - } + assert request_function_call.call_id == "request_123" + assert isinstance(request_function_call.arguments, dict) + assert request_function_call.arguments.get("data") is not None + data = request_function_call.arguments["data"] + assert data.get("type") == "request_info" + assert data.get("request_id") == "request_123" + assert data.get("source_executor_id") == "executor1" + assert deserialize_type(data.get("response_type")) is str + assert data.get("data") == {"target_agent": "helper", "reason": "overflow"} + + def test_process_request_info_event_passes_through_function_approval_request(self) -> None: + """If the event data is already a function approval request, it is forwarded unchanged. + + Tool-approval requests emitted by an inner agent surface as ``Content`` + objects with ``user_input_request=True``. ``WorkflowAgent`` must not + re-wrap these inside a synthesized ``request_info`` function call; + instead it should return the original content as-is so callers can + respond with a matching ``function_approval_response``. + """ + executor = SimpleExecutor(id="executor1", response_text="Response") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="Approval Passthrough Agent") + + approval_id = "approval-passthrough-1" + inner_function_call = Content.from_function_call( + call_id="tool-call-1", + name="delete_file", + arguments={"path": "/tmp/x"}, + ) + approval_request = Content.from_function_approval_request( + id=approval_id, + function_call=inner_function_call, + ) + event = WorkflowEvent.request_info( + request_id=approval_id, + source_executor_id="executor1", + request_data=approval_request, + response_type=Content, + ) + + result = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage] + + # The original FunctionApprovalRequestContent is returned as-is — same + # instance, with the original tool name preserved (NOT replaced by the + # synthesized REQUEST_INFO_FUNCTION_NAME). + assert result is approval_request + assert result.type == "function_approval_request" + assert result.id == approval_id + assert result.user_input_request is True + assert result.function_call is inner_function_call # type: ignore[attr-defined] + assert result.function_call.name == "delete_file" # type: ignore[attr-defined] + assert result.function_call.name != WorkflowAgent.REQUEST_INFO_FUNCTION_NAME # type: ignore[attr-defined] + + def test_extract_function_responses_passes_through_approval_response_approved(self) -> None: + """A function_approval_response with approved=True is keyed by content.id and forwarded as-is. + + After the refactor, ``WorkflowAgent`` no longer unwraps a synthesized + ``request_info`` function call from approval responses — the response + content is routed straight back to the workflow under its own ``id``, + which matches the pending request id surfaced by + ``_process_request_info_event``. + """ + executor = SimpleExecutor(id="executor1", response_text="Response") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="Approval Response Agent") + + approval_id = "approval-response-approved-1" + inner_function_call = Content.from_function_call( + call_id="tool-call-1", + name="delete_file", + arguments={"path": "/tmp/x"}, + ) + approval_request = Content.from_function_approval_request( + id=approval_id, + function_call=inner_function_call, + ) + approval_response = approval_request.to_function_approval_response(approved=True) # type: ignore[attr-defined] + message = Message(role="user", contents=[approval_response]) + + responses = agent._extract_function_responses([message]) # pyright: ignore[reportPrivateUsage] + + assert set(responses.keys()) == {approval_id} + assert responses[approval_id] is approval_response + assert responses[approval_id].approved is True # type: ignore[attr-defined] + + def test_extract_function_responses_passes_through_approval_response_denied(self) -> None: + """A function_approval_response with approved=False is forwarded the same way as an approval. + + Only the ``approved`` flag changes — routing back to the workflow is + identical for accept and reject paths. + """ + executor = SimpleExecutor(id="executor1", response_text="Response") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="Approval Response Agent") + + approval_id = "approval-response-denied-1" + inner_function_call = Content.from_function_call( + call_id="tool-call-2", + name="send_email", + arguments={"to": "alice@example.com"}, + ) + approval_request = Content.from_function_approval_request( + id=approval_id, + function_call=inner_function_call, + ) + approval_response = approval_request.to_function_approval_response(approved=False) # type: ignore[attr-defined] + message = Message(role="user", contents=[approval_response]) + + responses = agent._extract_function_responses([message]) # pyright: ignore[reportPrivateUsage] + + assert set(responses.keys()) == {approval_id} + assert responses[approval_id] is approval_response + assert responses[approval_id].approved is False # type: ignore[attr-defined] + + async def test_function_approval_request_flows_end_to_end_approved(self) -> None: + """End-to-end: an executor emits a function_approval_request, the agent + forwards it unchanged, and an ``approved=True`` response resumes the workflow. + + This exercises the full pass-through path: + ``ctx.request_info(approval_content, ...)`` -> ``WorkflowAgent`` surfaces + the original ``FunctionApprovalRequestContent`` -> caller responds with a + ``FunctionApprovalResponseContent`` -> ``WorkflowAgent`` routes it back + to the workflow which delivers it to the executor's ``@response_handler``. + """ + approval_id = "e2e-approval-1" + inner_function_call = Content.from_function_call( + call_id="tool-call-e2e-1", + name="delete_file", + arguments={"path": "/tmp/x"}, + ) + approval_request = Content.from_function_approval_request( + id=approval_id, + function_call=inner_function_call, + ) + + class ApprovalRequestingExecutor(Executor): + @handler + async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None: + await ctx.request_info(approval_request, Content, request_id=approval_id) + + @response_handler + async def handle_response( + self, + original_request: Content, + response: Content, + ctx: WorkflowContext[Never, AgentResponse], + ) -> None: + assert response.type == "function_approval_response" + assert response.id == approval_id # type: ignore[attr-defined] + approved = bool(response.approved) # type: ignore[attr-defined] + tool_name = original_request.function_call.name # type: ignore[attr-defined] + await ctx.yield_output( + AgentResponse( + messages=[ + Message( + role="assistant", + contents=[Content.from_text(text=f"{tool_name} approved={approved}")], + ) + ] + ) + ) + + executor = ApprovalRequestingExecutor(id="approval_requester") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="E2E Approval Agent") + + # First run: workflow pauses with the approval request. + first = await agent.run("please delete it") + assert isinstance(first, AgentResponse) + + forwarded = next( + ( + c + for m in first.messages + for c in m.contents + if c.type == "function_approval_request" and c.id == approval_id + ), + None, + ) + assert forwarded is approval_request, "Approval request must surface unchanged" + + pending = await workflow._runner_context.get_pending_request_info_events() + assert approval_id in pending + + # Respond with approved=True. + approval_response = approval_request.to_function_approval_response(approved=True) # type: ignore[attr-defined] + final = await agent.run(Message(role="user", contents=[approval_response])) + + assert isinstance(final, AgentResponse) + final_text = " ".join(m.text or "" for m in final.messages) + assert "delete_file approved=True" in final_text + + pending = await workflow._runner_context.get_pending_request_info_events() + assert approval_id not in pending + + async def test_function_approval_request_flows_end_to_end_denied(self) -> None: + """End-to-end denied path: ``approved=False`` is delivered to the executor's + response handler so the workflow can branch on the rejection.""" + approval_id = "e2e-approval-deny-1" + inner_function_call = Content.from_function_call( + call_id="tool-call-e2e-deny-1", + name="send_email", + arguments={"to": "alice@example.com"}, + ) + approval_request = Content.from_function_approval_request( + id=approval_id, + function_call=inner_function_call, + ) + + class ApprovalRequestingExecutor(Executor): + @handler + async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None: + await ctx.request_info(approval_request, Content, request_id=approval_id) + + @response_handler + async def handle_response( + self, + original_request: Content, + response: Content, + ctx: WorkflowContext[Never, AgentResponse], + ) -> None: + assert response.type == "function_approval_response" + assert response.id == approval_id # type: ignore[attr-defined] + approved = bool(response.approved) # type: ignore[attr-defined] + tool_name = original_request.function_call.name # type: ignore[attr-defined] + await ctx.yield_output( + AgentResponse( + messages=[ + Message( + role="assistant", + contents=[Content.from_text(text=f"{tool_name} approved={approved}")], + ) + ] + ) + ) + + executor = ApprovalRequestingExecutor(id="approval_requester_deny") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="E2E Approval Deny Agent") + + first = await agent.run("please send") + assert isinstance(first, AgentResponse) + forwarded = next( + ( + c + for m in first.messages + for c in m.contents + if c.type == "function_approval_request" and c.id == approval_id + ), + None, + ) + assert forwarded is approval_request + + # Respond with approved=False. + approval_response = approval_request.to_function_approval_response(approved=False) # type: ignore[attr-defined] + final = await agent.run(Message(role="user", contents=[approval_response])) + + assert isinstance(final, AgentResponse) + final_text = " ".join(m.text or "" for m in final.messages) + assert "send_email approved=False" in final_text + + pending = await workflow._runner_context.get_pending_request_info_events() + assert approval_id not in pending + + async def test_request_info_non_approval_flows_end_to_end(self) -> None: + """End-to-end: when request data is not a function approval content, the + agent surfaces a synthesized ``function_call`` (name=REQUEST_INFO_FUNCTION_NAME) + and routes a matching ``function_result`` back to the executor. + """ + + @dataclass + class HandoffRequest: + target_agent: str + reason: str + + captured: dict[str, Any] = {} + + class HandoffRequestingExecutor(Executor): + @handler + async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None: + await ctx.request_info( + HandoffRequest(target_agent="helper", reason="overflow"), + str, + ) + + @response_handler + async def handle_response( + self, + original_request: HandoffRequest, + response: str, + ctx: WorkflowContext[Never, AgentResponse], + ) -> None: + captured["original"] = original_request + captured["response"] = response + await ctx.yield_output( + AgentResponse( + messages=[ + Message( + role="assistant", + contents=[ + Content.from_text(text=f"handoff to {original_request.target_agent}: {response}") + ], + ) + ] + ) + ) + + executor = HandoffRequestingExecutor(id="handoff_requester") + workflow = WorkflowBuilder(start_executor=executor).build() + agent = WorkflowAgent(workflow=workflow, name="E2E Handoff Agent") + + # First run: workflow pauses with a synthesized request_info function_call. + first = await agent.run("start handoff") + assert isinstance(first, AgentResponse) + + function_call = next( + ( + c + for m in first.messages + for c in m.contents + if c.type == "function_call" and c.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME + ), + None, + ) + assert function_call is not None, "Expected a synthesized request_info function_call" + assert function_call.call_id is not None + assert isinstance(function_call.arguments, dict) + request_id = function_call.arguments["request_id"] + assert function_call.call_id == request_id + request_payload = function_call.arguments["data"] + assert request_payload.get("type") == "request_info" + assert request_payload.get("data") == {"target_agent": "helper", "reason": "overflow"} + + pending = await workflow._runner_context.get_pending_request_info_events() + assert request_id in pending + + # Respond with a function_result keyed by the call_id. + function_result = Content.from_function_result(call_id=request_id, result="ok-do-it") + final = await agent.run(Message(role="user", contents=[function_result])) + + assert isinstance(final, AgentResponse) + final_text = " ".join(m.text or "" for m in final.messages) + assert "handoff to helper: ok-do-it" in final_text + + # The executor's response handler received the original request and the response. + assert isinstance(captured.get("original"), HandoffRequest) + assert captured["original"].target_agent == "helper" + assert captured["response"] == "ok-do-it" + + pending = await workflow._runner_context.get_pending_request_info_events() + assert request_id not in pending def test_workflow_as_agent_method(self) -> None: """Test that Workflow.as_agent() creates a properly configured WorkflowAgent.""" 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 df31495559..878e54538e 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -618,7 +618,11 @@ class ResponsesHostServer(ResponsesAgentServerHost): checkpoint_storage=write_storage, ) - async for item in _to_outputs_for_messages(response_event_stream, response.messages): + async for item in _to_outputs_for_messages( + response_event_stream, + response.messages, + approval_storage=self._approval_storage, + ): yield item await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name) diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 759ef582a8..2aa0bccbb9 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -3520,6 +3520,7 @@ class _ToolApprovalWorkflowAgentMock(SupportsAgentRun): session: AgentSession | None = ..., **kwargs: Any, ) -> Awaitable[AgentResponse[Any]]: ... + @overload def run( self,