Confirmation Required MCP Toolkit

"""
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())

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())