External Tool Execution

"""
External Tool Execution
=============================

Human-in-the-Loop: Execute a tool call outside of the agent.
"""

import subprocess

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
from agno.utils import pprint


# We have to create a tool with the correct name, arguments and docstring for the agent to know what to call.
@tool(external_execution=True)
def execute_shell_command(command: str) -> str:
    """Execute a shell command.

    Args:
        command (str): The shell command to execute

    Returns:
        str: The output of the shell command
    """
    if command.startswith("ls"):
        return subprocess.check_output(command, shell=True).decode("utf-8")
    else:
        raise Exception(f"Unsupported command: {command}")


# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
    model=OpenAIResponses(id="gpt-5-mini"),
    tools=[execute_shell_command],
    markdown=True,
    db=SqliteDb(session_table="test_session", db_file="tmp/example.db"),
)

# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    run_response = agent.run("What files do I have in my current directory?")

    if run_response.is_paused:
        for requirement in run_response.active_requirements:
            if requirement.needs_external_execution:
                if requirement.tool_execution.tool_name == execute_shell_command.name:
                    print(
                        f"Executing {requirement.tool_execution.tool_name} with args {requirement.tool_execution.tool_args} externally"
                    )
                    # We execute the tool ourselves. You can also execute something completely external here.
                    result = execute_shell_command.entrypoint(
                        **requirement.tool_execution.tool_args
                    )  # type: ignore
                    # We have to set the result on the tool execution object so that the agent can continue
                    requirement.set_external_execution_result(result)

    run_response = agent.continue_run(
        run_id=run_response.run_id,
        requirements=run_response.requirements,
    )
    pprint.pprint_run_response(run_response)

    # Or for simple debug flow
    # agent.print_response("What files do I have in my current directory?")

from pathlib import Path
import subprocess

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
from agno.utils import pprint

DIRECTORY = Path.cwd()

@tool(external_execution=True)
def execute_shell_command(command: str) -> str:
    """List this example's working directory. The only accepted command is ls."""
    if command != "ls":
        raise ValueError("Only the exact command 'ls' is supported")
    return subprocess.run(
        ["ls"], cwd=DIRECTORY, shell=False, check=True,
        capture_output=True, text=True,
    ).stdout

agent = Agent(
    model=OpenAIResponses(id="gpt-5-mini"),
    tools=[execute_shell_command],
    db=SqliteDb(db_file="tmp/external_listing.db"),
    markdown=True,
)

if __name__ == "__main__":
    response = agent.run("Use the ls command to list the current directory.")
    while response.is_paused:
        for requirement in response.active_requirements:
            execution = requirement.tool_execution
            if not requirement.needs_external_execution or execution is None:
                raise RuntimeError("Unexpected requirement; do not continue it")
            if execution.tool_name != execute_shell_command.name:
                raise ValueError("Unsupported external tool")
            result = execute_shell_command.entrypoint(**(execution.tool_args or {}))
            requirement.set_external_execution_result(result)
        response = agent.continue_run(
            run_id=response.run_id, requirements=response.requirements,
        )
    pprint.pprint_run_response(response)