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 @tool(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?")
for requirement in run_response.active_requirements:
if requirement.is_external_tool_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")
# Execute the tool manually. You can execute any function or process here and use the tool_args as input.
result = execute_shell_command.entrypoint(**requirement.tool_execution.tool_args)
# Set the result on the tool execution object so that the agent can continue
requirement.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)
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 returned RunOutput will contain a list of requirement objects.
These requirement objects will contain the tool executions that need to run outside of the agent’s run.
You can find the tool related to each requirement in requirement.tool_execution. Each tool execution object contains:
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 requirements, execute the tools however you want, and set their results:
for requirement in run_response.active_requirements:
if requirement.is_external_tool_execution:
print(f"Tool: {requirement.tool_execution.tool_name}")
print(f"Args: {requirement.tool_execution.tool_args}")
# Execute your custom logic here
result = my_custom_execution(requirement.tool_execution.tool_args)
# Set the result so the agent can continue
requirement.external_execution_result = result
# After resolving the requirement, you can continue the run:
response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)
Important: You must resolve all external tool execution requirements before calling continue_run().
An external tool execution requirement is considered resolved when you set requirement.external_execution_result.Else Agno will raise a ValueError, letting you know that not all requirements have been resolved.
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?")
for requirement in run_response.active_requirements:
if requirement.is_external_tool_execution:
# Only list_dir will be here, get_env runs normally
if requirement.tool_execution.tool_name == "list_dir":
result = ShellTools().list_dir(**requirement.tool_execution.tool_args)
requirement.external_execution_result = result
# After resolving the requirement, you can continue the run:
response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)
@tool(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
for requirement in response.active_requirements:
if requirement.is_external_tool_execution:
if requirement.tool_execution.tool_name == "sensitive_database_query":
# Execute with your own DB connection and security checks
result = execute_safe_db_query(requirement.tool_execution.tool_args["query"])
requirement.external_execution_result = result
# After resolving the requirement, you can continue the run:
response = agent.continue_run(run_id=response.run_id, requirements=response.requirements)
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")
for requirement in run_response.active_requirements:
if requirement.is_external_tool_execution:
# Execute your async external logic
result = await my_async_external_service(requirement.tool_execution.tool_args)
requirement.external_execution_result = result
response = await agent.acontinue_run(run_id=run_response.run_id, requirements=run_response.requirements)
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 requirement in run_event.active_requirements:
if requirement.is_external_tool_execution:
# Execute externally
result = execute_tool_externally(requirement.tool_execution.tool_args)
requirement.external_execution_result = result
# Continue streaming
for response in agent.continue_run(
run_id=run_event.run_id,
requirements=run_event.requirements,
stream=True
):
print(response.content, end="")
else:
print(run_event.content, end="")
Best Practices
- Always set results: Make sure you set
requirement.external_execution_result for all requirements 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 @tool(external_execution=True) are mutually exclusive with @tool(requires_confirmation=True) and @tool(requires_user_input=True).A tool can only use one of these patterns at a time.
Usage Examples
Developer Resources
