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.
"""🤝 Human-in-the-Loop: Execute a tool call outside of the agentThis 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 agentRun `pip install openai agno` to install dependencies."""import asyncioimport subprocessfrom agno.agent import Agentfrom agno.db.sqlite import SqliteDbfrom agno.models.openai import OpenAIChatfrom 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())