External tool execution gives you complete control over when and how certain tools actually run. Instead of letting the agent execute the tool directly, it pauses and waits for you to handle the execution yourself. This is incredibly useful when you need:
- Enhanced security: Execute sensitive operations in a controlled environment
- External service calls: Integrate with services that require special handling
- Database operations: Run queries through your own connection management
- Custom execution logic: Add validation, logging, or rate limiting before execution
- Sandboxed environments: Execute potentially dangerous operations safely
How It Works
When you mark a tool with external_execution=True, your agent will:
- Pause execution when the tool is about to be called
- Set
is_paused to True on the run response
- Populate
tools_awaiting_external_execution with tools that need external handling
- Wait for you to execute the tool and set its result
- Continue execution once you call
continue_run() with the result
The key difference from other HITL patterns is that the agent never actually calls the function—you’re responsible for the entire execution.
import subprocess
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.utils import pprint
# 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
"""
return subprocess.check_output(command, shell=True).decode("utf-8")
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[execute_shell_command],
markdown=True,
)
run_response = agent.run("What files do I have in my current directory?")
if run_response.is_paused:
for tool in run_response.tools_awaiting_external_execution:
if tool.tool_name == execute_shell_command.name:
print(f"Executing {tool.tool_name} with args {tool.tool_args} externally")
# Execute the tool manually. You can execute any function or process here and use the tool_args as input.
result = execute_shell_command.entrypoint(**tool.tool_args)
# Set the result on the tool execution object so that the agent can continue
tool.result = result
run_response = agent.continue_run(run_id=run_response.run_id, updated_tools=run_response.tools)
pprint.pprint_run_response(run_response)
execute_shell_command but doesn’t actually execute it. Instead, it pauses and gives you the tool name and arguments. You then execute it yourself (or something completely different!) and provide the result back.
When a run is paused for external execution, the tools_awaiting_external_execution property contains all tools that need to be executed externally. Each tool has:
tool_name: The name of the tool that was calledtool_args: A dictionary of arguments the agent wants to pass to the toolexternal_execution_required: A boolean flag set to Trueresult: Where you set the execution result (initially None)
You can iterate through these tools, execute them however you want, and set their results:
if run_response.is_paused:
for tool in run_response.tools_awaiting_external_execution:
print(f"Tool: {tool.tool_name}")
print(f"Args: {tool.tool_args}")
# Execute your custom logic here
result = my_custom_execution(tool.tool_args)
# Set the result so the agent can continue
tool.result = result
# Continue the run with the updated tools
response = agent.continue_run(run_id=run_response.run_id, updated_tools=run_response.tools)
Important: You must set tool.result for all tools in tools_awaiting_external_execution before calling continue_run(). If you try to continue without setting results, Agno will raise a ValueError.
Toolkit, you can specify which tools require external execution using the external_execution_required_tools parameter:
from agno.tools.toolkit import Toolkit
import subprocess
class ShellTools(Toolkit):
def __init__(self, *args, **kwargs):
super().__init__(
tools=[self.list_dir, self.get_env],
external_execution_required_tools=["list_dir"], # Only this one needs external execution
*args,
**kwargs,
)
def list_dir(self, directory: str):
"""Lists the contents of a directory."""
return subprocess.check_output(f"ls {directory}", shell=True).decode("utf-8")
def get_env(self, var_name: str):
"""Gets an environment variable."""
import os
return os.getenv(var_name, "Not found")
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[ShellTools()],
markdown=True,
)
run_response = agent.run("What files are in my current directory and what's my PATH?")
if run_response.is_paused:
# Only list_dir will be here, get_env runs normally
for tool in run_response.tools_awaiting_external_execution:
if tool.tool_name == "list_dir":
result = ShellTools().list_dir(**tool.tool_args)
tool.result = result
run_response = agent.continue_run(run_id=run_response.run_id, updated_tools=run_response.tools)
external_execution=True will cause a pause:
@tool(external_execution=True)
def sensitive_database_query(query: str) -> str:
"""Execute a database query."""
pass
@tool
def safe_calculation(x: int, y: int) -> int:
"""Perform a safe calculation."""
return x + y
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[sensitive_database_query, safe_calculation],
markdown=True,
)
response = agent.run("Calculate 5 + 10 and query the users table")
# Agent will pause when it tries to call sensitive_database_query
# but safe_calculation executes normally
if response.is_paused:
for tool in response.tools_awaiting_external_execution:
if tool.tool_name == "sensitive_database_query":
# Execute with your own DB connection and security checks
result = execute_safe_db_query(tool.tool_args["query"])
tool.result = result
response = agent.continue_run(run_id=response.run_id, updated_tools=response.tools)
Async Support
External execution works seamlessly with async operations. Use arun() and acontinue_run() for async flows:
import asyncio
@tool(external_execution=True)
async def async_external_tool(data: str) -> str:
"""An async tool requiring external execution."""
pass
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[async_external_tool],
markdown=True,
)
async def main():
run_response = await agent.arun("Process some data")
if run_response.is_paused:
for tool in run_response.tools_awaiting_external_execution:
# Execute your async external logic
result = await my_async_external_service(tool.tool_args)
tool.result = result
response = await agent.acontinue_run(run_id=run_response.run_id, updated_tools=run_response.tools)
print(response.content)
asyncio.run(main())
Streaming Support
You can also use external execution with streaming responses:
for run_event in agent.run("What files are in my directory?", stream=True):
if run_event.is_paused:
for tool in run_event.tools_awaiting_external_execution:
# Execute externally
result = execute_tool_externally(tool)
tool.result = result
# Continue streaming
for response in agent.continue_run(
run_id=run_event.run_id,
updated_tools=run_event.tools,
stream=True
):
print(response.content, end="")
else:
print(run_event.content, end="")
Best Practices
- Always set results: Make sure you set
tool.result for all tools in tools_awaiting_external_execution before continuing
- Error handling: Wrap your external execution in try-catch blocks and provide meaningful error messages as results
- Security validation: Use external execution to add extra security checks before running sensitive operations
- Logging: Log all external executions for audit trails
- Timeouts: Consider adding timeouts to your external execution logic to prevent hanging
Remember that external execution tools marked with external_execution=True are mutually exclusive with requires_confirmation=True and requires_user_input=True. A tool can only use one of these patterns at a time.
Usage Examples
Developer Resources
