Human in the Loop Approval for Multi Agent Systems

[CantLooseToAMoose]

Question

Hey everyone,

I'm working with a multi-agent system and need some guidance on implementing human approval workflows for delegated tasks.

Scenario:
I have a coordinator agent that delegates tasks to specialized agents (e.g., a file management specialist). When the specialist needs to perform a sensitive operation like deleting a file, I want to require human approval before proceeding.

Current approach:

The specialist returns a DeferredToolRequest for the deletion
The coordinator handles this by raising an ApprovalRequired exception
After receiving human approval, I create DeferredToolResults and pass them back to the coordinator
The issue:
When I provide the DeferredToolResults to the coordinator, it seems to just call the specialized agent again, creating a loop rather than completing the approved action.

Question:
What's the correct pattern for handling human approval in delegated tool calls? How should I structure the flow so that once approval is granted, the original tool request gets executed properly?

Any insights or examples would be greatly appreciated!

TL;DR: How do I properly implement human approval for tools called by delegated agents without getting stuck in loops?

Additional Context

No response

[DouweM]

@CantLooseToAMoose In this case both the coordinator and the specialist agents need to be resumed with their own message history + DeferredToolResults.

So:

• When inside the coordinator's run_specialist tool result = specialist_agent.run() outputs DeferredToolRequest, you should store result.all_messages() and that DeferredToolRequest (or at least the tool call IDs) somewhere, possibly via deps, before raising ApprovalRequired for the coordinator agent.
• Then when result = coordinator_agent.run() outputs DeferredToolRequest, you should also store its result.all_messages() and that DeferredToolRequest and present the request to the user.
• Then when the user has approved, you pass the coordinator's message_history + DeferredToolResults(approvals={<tool_call_id>: True}) to coordinator_agent.run() which will result in the run_specialist tool being called again.
• Inside that tool, assuming it takes ctx: RunContext, you can check ctx.tool_call_approved to know if the current call has already been approved.
• If so, you should pass the specialist's message_history that you stored earlier + DeferredToolResults(approvals={<tool_call_id>: True}) to specialist_agent.run() to get it to pick up where it was interrupted.

You can use deps to store the specialist's message history + deferred tool requests/results, and also use deps to pass that in again.

Something like:

from dataclasses import dataclass, field

import logfire

from pydantic_ai import Agent, ApprovalRequired, DeferredToolRequests, DeferredToolResults, ModelMessage, RunContext

logfire.configure()
logfire.instrument_pydantic_ai()

specialist_agent = Agent('openai:gpt-5', output_type=[str, DeferredToolRequests])


@specialist_agent.tool_plain(requires_approval=True)
def get_answer(question: str) -> str:
    """Always use this tool to answer questions."""
    return 'The answer to the question is 42.'


@dataclass
class CoordinatorDeps:
    specialist_approval_requests: dict[str, tuple[list[ModelMessage], DeferredToolRequests]] = field(
        default_factory=dict
    )


coordinator_agent = Agent('openai:gpt-5', output_type=[str, DeferredToolRequests], deps_type=CoordinatorDeps)


@coordinator_agent.tool
def run_specialist(ctx: RunContext[CoordinatorDeps], question: str) -> str:
    """Always use the specialist to answer questions."""
    user_prompt: str | None = question
    message_history = []
    deferred_tool_results: DeferredToolResults | None = None
    if ctx.tool_call_approved and ctx.tool_call_id is not None:
        try:
            message_history, deferred_tool_requests = ctx.deps.specialist_approval_requests.pop(ctx.tool_call_id)
            deferred_tool_results = DeferredToolResults(
                approvals={call.tool_call_id: True for call in deferred_tool_requests.approvals}
            )
            user_prompt = None
        except KeyError:
            pass

    result = specialist_agent.run_sync(
        user_prompt, message_history=message_history, deferred_tool_results=deferred_tool_results
    )
    output = result.output
    if isinstance(output, DeferredToolRequests):
        assert ctx.tool_call_id is not None
        ctx.deps.specialist_approval_requests[ctx.tool_call_id] = (result.all_messages(), output)
        raise ApprovalRequired
    return output


deps = CoordinatorDeps()

user_prompt: str | None = 'What is the answer to the question of life, the universe, and everything?'
message_history = []
deferred_tool_results: DeferredToolResults | None = None
output: str | None = None
while True:
    result = coordinator_agent.run_sync(
        user_prompt,
        deps=deps,
        message_history=message_history,
        deferred_tool_results=deferred_tool_results,
    )

    if isinstance(result.output, DeferredToolRequests):
        user_prompt = None
        message_history = result.all_messages()
        # Get approval from user here:
        deferred_tool_results = DeferredToolResults(
            approvals={call.tool_call_id: True for call in result.output.approvals}
        )
    else:
        output = result.output
        break

print(output)

[dav04]

Thanks for this example.
When you say "# Get approval from user here:", the best solution is to integrate an agent.run to ask user a confirmation? Or do you think it is possible asking with a direct method?
In my case, I'm using agent in a chat with StreamingResponse of FastAPI.

[DouweM]

@dav04 Are you using one of the frontend protocols from https://ai.pydantic.dev/ui/overview/, or something custom?

Typically you'd consider this agent run complete once it returns DeferredToolRequests, then send some event to the frontend asking the user for approval, and then once the frontend receives the approval, it'll send a new request that you can use to continue the agent run, passing in the message_history.

You'd need to store the deps.specialist_approval_requests somewhere (possibly by sending it to the frontend and receiving it again on the next run request) so that the next run can use them.

[dav04]

@DouweM I have started from your example of chat app that doesn't have a protocol defined.
Do you think it is better to use a protocol?

So, if I want to get a feedback from the user, I have to temporary save the approval state on the frontend to return it again with the response of the user and with message history (actually saved on DB using new_messages_json).

[DouweM]

Do you think it is better to use a protocol?

@dav04 Yeah I recommend using a chat app frontend framework like CopilotKit (AG-UI) or Vercel AI Elements (Vercel AI Data Stream Protocol)

So, if I want to get a feedback from the user, I have to temporary save the approval state on the frontend to return it again with the response of the user and with message history (actually saved on DB using new_messages_json).

Yep, store it somewhere so you can get it on the next run request, that can be by storing it on the frontend and sending it back, or storing it in the DB

[github-actions]

This issue is stale, and will be closed in 3 days if no reply is received.

[DouweM]

Note that I intend for this to be handled transparently once we get have native handoffs: #1978 (comment).

[phiweger]

#1978 (comment)

When do you think this is ready?

[DouweM]

@phiweger It's currently in the milestone for next month

[bashiryounis]

I want to share how I handled HITL approval in a multi-agent setup (supervisor + child agents), and a detail that I think is important for others implementing this pattern:

Important Note: You MUST persist the child agent's memory

To correctly resume the child agent after the user approves, you must store the child’s message history somewhere (DB, Redis, or your frontend) before raising ApprovalRequired.

Otherwise, when the approval returns, the supervisor won't know:

which child tool call was pending

the last tool_call_id

the full conversation context needed to resume the agent

In my implementation, I store result.all_messages() before raising ApprovalRequired.
Then, on the next run, I load this memory and inspect the last message to recover the pending tool call.

async def handoff_task(
    ctx: RunContext[AgentDeps],
    task: str,
    intent: Literal["booking", "search", "support"],
    context: str | None = None,
) -> str | InvoiceOutput:

    logger.info(f"[Handoff] Routing to {intent} agent. Supervisor Approved? {ctx.tool_call_approved}")
    
    user_id = ctx.deps.user_id
    chat_id = ctx.deps.chat_id

    # 1. Select child agent
    agent_map = {
        "booking": booking_agent,
        "search": search_agent,
        "support": customer_service_agent,
    }
    child_agent = agent_map[intent]

    # 2. Load previously stored child memory (needed to resume)
    child_memory = get_memories(user_id=user_id, chat_id=chat_id, agent_name=intent) or []

    # 3. Resume logic if supervisor just got approval
    deferred_results_for_child = None
    if ctx.tool_call_approved:
        pending_child_tool_id = None

        # Look for last tool call in stored memory
        if child_memory and isinstance(child_memory[-1], ModelResponse):
            for part in child_memory[-1].parts:
                if isinstance(part, ToolCallPart):
                    pending_child_tool_id = part.tool_call_id
                    break

        if pending_child_tool_id:
            deferred_results_for_child = DeferredToolResults()
            deferred_results_for_child.approvals[pending_child_tool_id] = True
            logger.info(f"[Handoff] Resuming child tool call={pending_child_tool_id}")

    # Build prompt
    prompt = f"{context}\n\nCurrent task: {task}" if context else task

    run_kwargs = {
        "deps": ctx.deps,
        "message_history": child_memory,
    }

    if deferred_results_for_child:
        # Resume from HITL approval
        run_kwargs["deferred_tool_results"] = deferred_results_for_child
        prompt_arg = None
    else:
        # New run
        prompt_arg = prompt

    result = await child_agent.run(prompt_arg, **run_kwargs)

    # 4. Child requested approval: bubble it up
    if isinstance(result.output, DeferredToolRequests):
        approval = result.output.approvals[0]

        # Persist child state BEFORE raising ApprovalRequired
        save_pydanticai_memory.delay(
            user_id, chat_id, to_jsonable_python(result.all_messages()), agent_name=intent
        )

        raise ApprovalRequired(
            metadata=result.output.metadata.get(approval.tool_call_id)
        )

    # 5. On success, save updated child memory
    save_pydanticai_memory.delay(
        user_id, chat_id, to_jsonable_python(result.all_messages()), agent_name=intent
    )

    return result.output

With this approach:

The supervisor does not loop
The child tool resumes exactly where it left off
HITL approval works cleanly end-to-end
Linked child & supervisor memory is preserved properly

[bashiryounis]

I'm trying to integrate Human-in-the-Loop with a single PydanticAI agent wrapped in TemporalAgent:

booking_agent = Agent(
    model,
    tools=[create_booking_invoice],  # Uses @agent.tool(requires_approval=True)
    output_type=[str, DeferredToolRequests],
)

temporal_booking_agent = TemporalAgent(booking_agent)

Problem: When create_booking_invoice raises ApprovalRequired, the agent returns DeferredToolRequests. But I don't know how to:
Return DeferredToolRequests from the workflow to the API layer
Resume the workflow with DeferredToolResults after user approval
Current workflow:

@workflow.defn
class BookingWorkflow:
    @workflow.run
    async def run(self, input: WorkflowInput) -> WorkflowOutput:
        result = await temporal_booking_agent.run(input.user_message, deps=deps)
        
        # How to handle DeferredToolRequests here?
        # How to pause workflow and wait for approval signal?
        
        return WorkflowOutput(ai=result.output)

What's the recommended pattern for HITL approval flows with TemporalAgent? Should I use Temporal signals to pause/resume, or is there built-in support?