Python [BREAKING]: support magentic agent tool call approvals and plan stalling HITL behavior (#2569)

* Provide way for HITL with magentic

* support tool call approvals and hitl stall replan

* human plan intervention sample

* Clean up

* Improve loging

* updates

Author: moonbox3

python/packages/core/agent_framework/_workflows/__init__.py

@@ -74,10 +74,15 @@
     ORCH_MSG_KIND_USER_TASK,
     MagenticBuilder,
     MagenticContext,
+    MagenticHumanInputRequest,
+    MagenticHumanInterventionDecision,
+    MagenticHumanInterventionKind,
+    MagenticHumanInterventionReply,
+    MagenticHumanInterventionRequest,
     MagenticManagerBase,
-    MagenticPlanReviewDecision,
-    MagenticPlanReviewReply,
-    MagenticPlanReviewRequest,
+    MagenticStallInterventionDecision,
+    MagenticStallInterventionReply,
+    MagenticStallInterventionRequest,
     StandardMagenticManager,
 )
 from ._orchestration_state import OrchestrationState
@@ -144,10 +149,15 @@
     "InProcRunnerContext",
     "MagenticBuilder",
     "MagenticContext",
+    "MagenticHumanInputRequest",
+    "MagenticHumanInterventionDecision",
+    "MagenticHumanInterventionKind",
+    "MagenticHumanInterventionReply",
+    "MagenticHumanInterventionRequest",
     "MagenticManagerBase",
-    "MagenticPlanReviewDecision",
-    "MagenticPlanReviewReply",
-    "MagenticPlanReviewRequest",
+    "MagenticStallInterventionDecision",
+    "MagenticStallInterventionReply",
+    "MagenticStallInterventionRequest",
     "ManagerDirectiveModel",
     "ManagerSelectionRequest",
     "ManagerSelectionResponse",

python/packages/core/agent_framework/_workflows/_magentic.py

@@ -12,16 +12,13 @@
     AgentRunResponseUpdate,
     AgentRunUpdateEvent,
     BaseAgent,
-    ChatClientProtocol,
     ChatMessage,
-    ChatResponse,
-    ChatResponseUpdate,
     Executor,
     MagenticBuilder,
+    MagenticHumanInterventionDecision,
+    MagenticHumanInterventionReply,
+    MagenticHumanInterventionRequest,
     MagenticManagerBase,
-    MagenticPlanReviewDecision,
-    MagenticPlanReviewReply,
-    MagenticPlanReviewRequest,
     RequestInfoEvent,
     Role,
     TextContent,
@@ -59,21 +56,25 @@ def test_magentic_start_message_from_string():
     assert msg.task.text == "Do the thing"
 
 
-def test_plan_review_request_defaults_and_reply_variants():
-    req = MagenticPlanReviewRequest()  # defaults provided by dataclass
+def test_human_intervention_request_defaults_and_reply_variants():
+    from agent_framework._workflows._magentic import MagenticHumanInterventionKind
+
+    req = MagenticHumanInterventionRequest(kind=MagenticHumanInterventionKind.PLAN_REVIEW)
     assert hasattr(req, "request_id")
     assert req.task_text == "" and req.facts_text == "" and req.plan_text == ""
     assert isinstance(req.round_index, int) and req.round_index == 0
 
     # Replies: approve, revise with comments, revise with edited text
-    approve = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)
-    revise_comments = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.REVISE, comments="Tighten scope")
-    revise_text = MagenticPlanReviewReply(
-        decision=MagenticPlanReviewDecision.REVISE,
+    approve = MagenticHumanInterventionReply(decision=MagenticHumanInterventionDecision.APPROVE)
+    revise_comments = MagenticHumanInterventionReply(
+        decision=MagenticHumanInterventionDecision.REVISE, comments="Tighten scope"
+    )
+    revise_text = MagenticHumanInterventionReply(
+        decision=MagenticHumanInterventionDecision.REVISE,
         edited_plan_text="- Step 1\n- Step 2",
     )
 
-    assert approve.decision == MagenticPlanReviewDecision.APPROVE
+    assert approve.decision == MagenticHumanInterventionDecision.APPROVE
     assert revise_comments.comments == "Tighten scope"
     assert revise_text.edited_plan_text is not None and revise_text.edited_plan_text.startswith("- Step 1")
 
@@ -220,15 +221,14 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
 
     req_event: RequestInfoEvent | None = None
     async for ev in wf.run_stream("do work"):
-        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
+        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticHumanInterventionRequest:
             req_event = ev
     assert req_event is not None
 
     completed = False
     output: list[ChatMessage] | None = None
-    async for ev in wf.send_responses_streaming(
-        responses={req_event.request_id: MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)}
-    ):
+    reply = MagenticHumanInterventionReply(decision=MagenticHumanInterventionDecision.APPROVE)
+    async for ev in wf.send_responses_streaming(responses={req_event.request_id: reply}):
         if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
             completed = True
         elif isinstance(ev, WorkflowOutputEvent):
@@ -265,7 +265,7 @@ async def replan(self, magentic_context: MagenticContext) -> ChatMessage:  # typ
     # Wait for the initial plan review request
     req_event: RequestInfoEvent | None = None
     async for ev in wf.run_stream("do work"):
-        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
+        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticHumanInterventionRequest:
             req_event = ev
     assert req_event is not None
 
@@ -274,13 +274,13 @@ async def replan(self, magentic_context: MagenticContext) -> ChatMessage:  # typ
     completed = False
     async for ev in wf.send_responses_streaming(
         responses={
-            req_event.request_id: MagenticPlanReviewReply(
-                decision=MagenticPlanReviewDecision.APPROVE,
+            req_event.request_id: MagenticHumanInterventionReply(
+                decision=MagenticHumanInterventionDecision.APPROVE,
                 comments="Looks good; consider Z",
             )
         }
     ):
-        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
+        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticHumanInterventionRequest:
             saw_second_review = True
         if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
             completed = True
@@ -338,7 +338,7 @@ async def test_magentic_checkpoint_resume_round_trip():
     task_text = "checkpoint task"
     req_event: RequestInfoEvent | None = None
     async for ev in wf.run_stream(task_text):
-        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
+        if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticHumanInterventionRequest:
             req_event = ev
     assert req_event is not None
 
@@ -359,13 +359,13 @@ async def test_magentic_checkpoint_resume_round_trip():
 
     orchestrator = next(exec for exec in wf_resume.executors.values() if isinstance(exec, MagenticOrchestratorExecutor))
 
-    reply = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)
+    reply = MagenticHumanInterventionReply(decision=MagenticHumanInterventionDecision.APPROVE)
     completed: WorkflowOutputEvent | None = None
     req_event = None
     async for event in wf_resume.run_stream(
         resume_checkpoint.checkpoint_id,
     ):
-        if isinstance(event, RequestInfoEvent) and event.request_type is MagenticPlanReviewRequest:
+        if isinstance(event, RequestInfoEvent) and event.request_type is MagenticHumanInterventionRequest:
             req_event = event
     assert req_event is not None
 
@@ -430,25 +430,33 @@ async def test_magentic_agent_executor_on_checkpoint_save_and_restore_roundtrip(
 from agent_framework import StandardMagenticManager  # noqa: E402
 
 
-class _StubChatClient(ChatClientProtocol):
-    @property
-    def additional_properties(self) -> dict[str, Any]:
-        """Get additional properties associated with the client."""
-        return {}
-
-    async def get_response(self, messages, **kwargs):  # type: ignore[override]
-        return ChatResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="ok")])
-
-    def get_streaming_response(self, messages, **kwargs) -> AsyncIterable[ChatResponseUpdate]:  # type: ignore[override]
-        async def _gen():
-            if False:
-                yield ChatResponseUpdate()  # pragma: no cover
+class _StubManagerAgent(BaseAgent):
+    """Stub agent for testing StandardMagenticManager."""
+
+    async def run(
+        self,
+        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        *,
+        thread: Any = None,
+        **kwargs: Any,
+    ) -> AgentRunResponse:
+        return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="ok")])
+
+    def run_stream(
+        self,
+        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        *,
+        thread: Any = None,
+        **kwargs: Any,
+    ) -> AsyncIterable[AgentRunResponseUpdate]:
+        async def _gen() -> AsyncIterable[AgentRunResponseUpdate]:
+            yield AgentRunResponseUpdate(message_deltas=[ChatMessage(role=Role.ASSISTANT, text="ok")])
 
         return _gen()
 
 
 async def test_standard_manager_plan_and_replan_via_complete_monkeypatch():
-    mgr = StandardMagenticManager(chat_client=_StubChatClient())
+    mgr = StandardMagenticManager(agent=_StubManagerAgent())
 
     async def fake_complete_plan(messages: list[ChatMessage], **kwargs: Any) -> ChatMessage:
         # Return a different response depending on call order length
@@ -481,7 +489,7 @@ async def fake_complete_replan(messages: list[ChatMessage], **kwargs: Any) -> Ch
 
 
 async def test_standard_manager_progress_ledger_success_and_error():
-    mgr = StandardMagenticManager(chat_client=_StubChatClient())
+    mgr = StandardMagenticManager(agent=_StubManagerAgent())
     ctx = MagenticContext(
         task=ChatMessage(role=Role.USER, text="task"),
         participant_descriptions={"alice": "desc"},
@@ -718,7 +726,7 @@ async def test_magentic_checkpoint_resume_rejects_participant_renames():
 
     req_event: RequestInfoEvent | None = None
     async for event in workflow.run_stream("task"):
-        if isinstance(event, RequestInfoEvent) and event.request_type is MagenticPlanReviewRequest:
+        if isinstance(event, RequestInfoEvent) and event.request_type is MagenticHumanInterventionRequest:
             req_event = event
 
     assert req_event is not None

python/packages/core/tests/workflow/test_magentic.py

@@ -105,6 +105,8 @@ For additional observability samples in Agent Framework, see the [observability
 | Handoff (Return-to-Previous) | [orchestration/handoff_return_to_previous.py](./orchestration/handoff_return_to_previous.py) | Return-to-previous routing: after user input, routes back to the previous specialist instead of coordinator using `.enable_return_to_previous()` |
 | Magentic Workflow (Multi-Agent) | [orchestration/magentic.py](./orchestration/magentic.py) | Orchestrate multiple agents with Magentic manager and streaming |
 | Magentic + Human Plan Review | [orchestration/magentic_human_plan_update.py](./orchestration/magentic_human_plan_update.py) | Human reviews/updates the plan before execution |
+| Magentic + Human Stall Intervention | [orchestration/magentic_human_replan.py](./orchestration/magentic_human_replan.py) | Human intervenes when workflow stalls with `with_human_input_on_stall()` |
+| Magentic + Agent Clarification | [orchestration/magentic_agent_clarification.py](./orchestration/magentic_agent_clarification.py) | Agents ask clarifying questions via `ask_user` tool with `@ai_function(approval_mode="always_require")` |
 | Magentic + Checkpoint Resume | [orchestration/magentic_checkpoint.py](./orchestration/magentic_checkpoint.py) | Resume Magentic orchestration from saved checkpoints |
 | Sequential Orchestration (Agents) | [orchestration/sequential_agents.py](./orchestration/sequential_agents.py) | Chain agents sequentially with shared conversation context |
 | Sequential Orchestration (Custom Executor) | [orchestration/sequential_custom_executors.py](./orchestration/sequential_custom_executors.py) | Mix agents with a summarizer that appends a compact summary |

python/samples/getting_started/workflows/README.md

@@ -48,13 +48,21 @@ async def main() -> None:
         tools=HostedCodeInterpreterTool(),
     )
 
+    # Create a manager agent for orchestration
+    manager_agent = ChatAgent(
+        name="MagenticManager",
+        description="Orchestrator that coordinates the research and coding workflow",
+        instructions="You coordinate a team to complete complex tasks efficiently.",
+        chat_client=OpenAIChatClient(),
+    )
+
     print("\nBuilding Magentic Workflow...")
 
     workflow = (
         MagenticBuilder()
         .participants(researcher=researcher_agent, coder=coder_agent)
         .with_standard_manager(
-            chat_client=OpenAIChatClient(),
+            agent=manager_agent,
             max_round_count=10,
             max_stall_count=3,
             max_reset_count=2,

python/samples/getting_started/workflows/agents/magentic_workflow_as_agent.py

@@ -65,6 +65,14 @@ async def main() -> None:
         tools=HostedCodeInterpreterTool(),
     )
 
+    # Create a manager agent for orchestration
+    manager_agent = ChatAgent(
+        name="MagenticManager",
+        description="Orchestrator that coordinates the research and coding workflow",
+        instructions="You coordinate a team to complete complex tasks efficiently.",
+        chat_client=OpenAIChatClient(),
+    )
+
     print("\nBuilding Magentic Workflow...")
 
     # State used by on_agent_stream callback
@@ -75,7 +83,7 @@ async def main() -> None:
         MagenticBuilder()
         .participants(researcher=researcher_agent, coder=coder_agent)
         .with_standard_manager(
-            chat_client=OpenAIChatClient(),
+            agent=manager_agent,
             max_round_count=10,
             max_stall_count=3,
             max_reset_count=2,