# Confirmation Required MCP Toolkit (/examples/agents/human-in-the-loop/confirmation-required-mcp-toolkit) ```python title="confirmation_required_mcp_toolkit.py" """ Confirmation Required MCP Toolkit ============================= Human-in-the-Loop: Adding User Confirmation to Tool Calls with MCP Servers. """ import asyncio from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.tools.mcp import MCPTools from rich.console import Console from rich.prompt import Prompt console = Console() # --------------------------------------------------------------------------- # Create Agent # --------------------------------------------------------------------------- mcp_tools = MCPTools( transport="streamable-http", url="https://docs.agno.com/mcp", requires_confirmation_tools=["SearchAgno"], # Note: Tool names are case-sensitive ) agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools], markdown=True, db=SqliteDb(db_file="tmp/confirmation_required_toolkit.db"), ) # --------------------------------------------------------------------------- # Run Agent # --------------------------------------------------------------------------- async def main(): async for run_event in agent.arun("What is Agno?", stream=True): if run_event.is_paused: # Handle confirmation requirements for requirement in run_event.active_requirements: if requirement.needs_confirmation: # Ask for confirmation console.print( f"Tool name [bold blue]{requirement.tool_execution.tool_name}({requirement.tool_execution.tool_args})[/] requires confirmation." ) message = ( Prompt.ask( "Do you want to continue?", choices=["y", "n"], default="y" ) .strip() .lower() ) if message == "n": requirement.reject() else: requirement.confirm() # Continue the run after handling all confirmations async for resp in agent.acontinue_run( run_id=run_event.run_id, requirements=run_event.requirements, stream=True, ): if resp.content: print(resp.content, end="") else: # Not paused - print the streaming content if run_event.content: print(run_event.content, end="") print() # Final newline if __name__ == "__main__": asyncio.run(main()) ``` ### Current runner [#current-runner] For the Docs Agent v2 server, use `search_docs` as the exact confirmation-gated tool name. The source above targets the older `SearchAgno` surface. Set `DOCS_MCP_URL` to the `/mcp` endpoint of your deployed v2 server; the legacy URL is not evidence that a deployment has migrated. Use this complete replacement. It displays content events, handles each pause after consuming its stream, and continues only after the pending confirmations are answered. ```python title="confirmation_required_mcp_toolkit.py" import asyncio from os import environ from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.tools.mcp import MCPTools from rich.prompt import Prompt async def main(): async with MCPTools( transport="streamable-http", url=environ["DOCS_MCP_URL"], requires_confirmation_tools=["search_docs"], ) as mcp_tools: agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools], db=SqliteDb(db_file="tmp/confirmation_required_toolkit.db"), markdown=True, ) stream = agent.arun( "Use search_docs to find how to create an Agno Agent.", stream=True, stream_events=True, ) while True: paused = None async for event in stream: if event.event == "RunContent" and event.content: print(event.content, end="", flush=True) elif event.event == "RunPaused": paused = event elif event.event == "RunError": print(f"\nRun failed: {event.content}") if paused is None: break for requirement in paused.active_requirements: if not requirement.needs_confirmation: raise RuntimeError("Unexpected requirement; do not continue it") execution = requirement.tool_execution choice = Prompt.ask( f"Allow {execution.tool_name}({execution.tool_args})?", choices=["y", "n"], default="n", ) if choice == "y": requirement.confirm() else: requirement.reject() stream = agent.acontinue_run( run_id=paused.run_id, requirements=paused.requirements, stream=True, stream_events=True, ) print() if __name__ == "__main__": asyncio.run(main()) ``` The model chooses whether to call a tool. Only `search_docs` is confirmation-gated here; configure other server tool names separately if they also need approval. ## Run the Example [#run-the-example] Mac Windows ```bash uv venv --python 3.12 source .venv/bin/activate ``` ```bash uv venv --python 3.12 .venv\Scripts\activate ``` ```bash uv pip install -U "agno[mcp]" openai sqlalchemy ``` Mac/Linux Windows ```bash export OPENAI_API_KEY="your_openai_api_key_here" ``` ```bash $Env:OPENAI_API_KEY="your_openai_api_key_here" ``` Replace the placeholder with your deployed v2 server's MCP URL. Mac/Linux Windows ```bash export DOCS_MCP_URL="https://your-docs-agent-host/mcp" ``` ```powershell $Env:DOCS_MCP_URL="https://your-docs-agent-host/mcp" ``` Save the complete current runner above as `confirmation_required_mcp_toolkit.py`, then run: ```bash python confirmation_required_mcp_toolkit.py ``` Full source: [cookbook/02\_agents/10\_human\_in\_the\_loop/confirmation\_required\_mcp\_toolkit.py](https://github.com/agno-agi/agno/blob/8f36eaf2d18e91afa7b327eec66a3cd3685dcb87/cookbook/02_agents/10_human_in_the_loop/confirmation_required_mcp_toolkit.py)