External Tool Execution Stream Async

This example demonstrates how to execute tools outside of the agent using external tool execution with async streaming responses. It shows how to handle external tool execution in an asynchronous environment while maintaining real-time streaming.

Code

external_tool_execution_stream_async.py

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

This example shows how to implement human-in-the-loop functionality in your Agno tools.
It shows how to:
- Use external tool execution to execute a tool call outside of the agent

Run `pip install openai agno` to install dependencies.
"""

import asyncio
import subprocess

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.tools import tool


# 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}")


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


async def main():
    async for run_event in agent.arun(
        "What files do I have in my current directory?", stream=True
    ):
        if run_event.is_paused:
            for tool in run_event.tools_awaiting_external_execution:  # type: ignore
                if tool.tool_name == execute_shell_command.name:
                    print(
                        f"Executing {tool.tool_name} with args {tool.tool_args} externally"
                    )
                    # We execute the tool ourselves. You can also execute something completely external here.
                    result = execute_shell_command.entrypoint(**tool.tool_args)  # type: ignore
                    # We have to set the result on the tool execution object so that the agent can continue
                    tool.result = result

            async for resp in agent.acontinue_run(  # type: ignore
                run_id=run_event.run_id,
                updated_tools=run_event.tools,
                stream=True,
            ):
                print(resp.content, end="")
        else:
            print(run_event.content, end="")

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


if __name__ == "__main__":
    asyncio.run(main())

Usage

1. Create a virtual environment

   Open the Terminal and create a python virtual environment.

   Mac

   python3 -m venv .venv
   source .venv/bin/activate
   
2. Install libraries

   pip install -U agno openai
   
3. Export your OpenAI API key

   Mac/Linux

     export OPENAI_API_KEY="your_openai_api_key_here"
   
4. Create a Python file

   Create a Python file and add the above code.

   touch external_tool_execution_stream_async.py
   
5. Run Agent

   Mac

   python external_tool_execution_stream_async.py
   
6. Find All Cookbooks

   Explore all the available cookbooks in the Agno repository. Click the link below to view the code on GitHub:

   Agno Cookbooks on GitHub