External Tool Execution

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. Use it 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

Setup

Create and activate a virtual environment, then install the dependencies and set your key:

uv pip install -U agno openai sqlalchemy
export OPENAI_API_KEY="your_openai_api_key"
mkdir -p tmp

The examples use a writable SQLite database for continuation by run ID. Retain both run_id and session_id and use the same database when resuming in a new process. Application code must authorize the caller's access to that session and its pending requirements.

How It Works

When you mark a tool with @tool(external_execution=True), your agent will:

1. Pause execution when the tool is about to be called
2. Set is_paused to True on the run response
3. Populate tools_awaiting_external_execution with tools that need external handling
4. Wait for you to execute the tool and set its result
5. 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.

Example

This example exposes a listing of the current directory. The application checks the requested command before invoking it; external execution alone does not validate arguments or sandbox code.

from pathlib import Path

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

@tool(external_execution=True)
def list_current_directory() -> str:
    """List file and directory names in the current directory."""
    return "\n".join(sorted(path.name for path in Path.cwd().iterdir()))

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[list_current_directory],
    db=SqliteDb(db_file="tmp/external-execution.db"),
)

def resolve_external_requirements(requirements):
    for requirement in requirements:
        if requirement.needs_external_execution:
            call = requirement.tool_execution
            if call.tool_name != list_current_directory.name or call.tool_args:
                raise ValueError("Unexpected external tool or arguments")
            result = list_current_directory.entrypoint()
            requirement.set_external_execution_result(result)

run_response = agent.run("List the files in the current directory")
while run_response.is_paused:
    resolve_external_requirements(run_response.active_requirements)
    run_response = agent.continue_run(
        run_id=run_response.run_id,
        session_id=run_response.session_id,
        requirements=run_response.requirements,
    )
print(run_response.status, run_response.content)

Agno pauses before executing the function. Your application receives the proposed tool call, invokes the permitted operation, then attaches its result with set_external_execution_result().

Understanding External Tool Execution Requirements

Read each pending call from requirement.tool_execution: tool_name identifies the function and tool_args contains its proposed arguments. Call requirement.set_external_execution_result(result) for every active external requirement before continuing. Unresolved external requirements raise ValueError on continuation.

For a later continuation, load the run from the same database using both its run_id and session_id. For an in-process, non-streaming continuation, you can instead pass the complete paused output as agent.continue_run(run_response=run_response) after resolving its requirements.

Using Toolkits with External Execution

For a toolkit, set external_execution_required_tools to the names requiring external handling:

from agno.tools.toolkit import Toolkit

class DirectoryTools(Toolkit):
    def __init__(self):
        super().__init__(
            tools=[self.list_directory, self.describe_directory],
            external_execution_required_tools=["list_directory"],
        )

    def list_directory(self) -> str:
        """List names in the current directory."""
        return "\n".join(sorted(path.name for path in Path.cwd().iterdir()))

    def describe_directory(self) -> str:
        """Describe the scope of the directory listing tool."""
        return "The listing contains names from the application's current directory."

Use DirectoryTools() in an agent's tools list. When a list_directory requirement arrives, check its name and arguments, call the toolkit's method externally, attach the result, and continue as above. describe_directory executes inside the agent's ordinary tool loop.

Mixed Tool Scenarios

Regular tools execute normally; only tools marked for external execution pause. The toolkit above demonstrates both kinds. For a database or service integration, validate the proposed operation and caller permissions in your application's dispatcher before executing it. Return a meaningful failure result if the external operation fails; do not report success for an operation that did not complete.

Async Support

Reuse the definitions above and run pause and continuation in one event loop. This fragment moves the directory listing to a worker thread:

import asyncio

async def main():
    response = await agent.arun("List the files in the current directory")
    while response.is_paused:
        await asyncio.to_thread(resolve_external_requirements, response.active_requirements)
        response = await agent.acontinue_run(
            run_id=response.run_id,
            session_id=response.session_id,
            requirements=response.requirements,
        )
    print(response.status, response.content)

asyncio.run(main())

Streaming Support

Reuse the agent and resolver above. Consume each stream completely, then resolve a pause and consume the continuation. Repeat if another pause occurs:

stream = agent.run("List the files in the current directory", stream=True)
while True:
    paused = None
    for event in stream:
        if event.is_paused:
            paused = event
        elif getattr(event, "content", None):
            print(event.content, end="")
    if paused is None:
        break
    resolve_external_requirements(paused.active_requirements)
    stream = agent.continue_run(
        run_id=paused.run_id,
        session_id=paused.session_id,
        requirements=paused.requirements,
        stream=True,
    )

Best Practices

Resolve every external requirement, validate proposed arguments, and apply your application's authorization, timeouts, and audit logging around the actual operation. An external-execution flag only transfers execution to your application; it does not supply those controls.

Usage Examples