External Tool Execution

Execute tools outside of the agent's control for enhanced security and flexibility.

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:

  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.

1import subprocess
2
3from kern.agent import Agent
4from kern.models.openai import OpenAIResponses
5from kern.tools import tool
6from kern.utils import pprint
7
8
9# Create a tool with the correct name, arguments and docstring for the agent to know what to call.
10@tool(external_execution=True)
11def execute_shell_command(command: str) -> str:
12 """Execute a shell command.
13
14 Args:
15 command (str): The shell command to execute
16
17 Returns:
18 str: The output of the shell command
19 """
20 return subprocess.check_output(command, shell=True).decode("utf-8")
21
22
23agent = Agent(
24 model=OpenAIResponses(id="gpt-5.2"),
25 tools=[execute_shell_command],
26 markdown=True,
27)
28
29run_response = agent.run("What files do I have in my current directory?")
30
31for requirement in run_response.active_requirements:
32 if requirement.is_external_tool_execution:
33 if requirement.tool_execution.tool_name == execute_shell_command.name:
34 print(f"Executing {requirement.tool_execution.tool_name} with args {requirement.tool_execution.tool_args} externally")
35
36 # Execute the tool manually. You can execute any function or process here and use the tool_args as input.
37 result = execute_shell_command.entrypoint(**requirement.tool_execution.tool_args)
38
39 # Set the result on the tool execution object so that the agent can continue
40 requirement.external_execution_result = result
41
42run_response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)
43pprint.pprint_run_response(run_response)

In this example, the agent identifies that it needs to run 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.

Understanding External Tool Execution Requirements

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 called
  • tool_args: A dictionary of arguments the agent wants to pass to the tool
  • external_execution_required: A boolean flag set to True
  • result: Where you set the execution result (initially None)

You can iterate through these requirements, execute the tools however you want, and set their results:

1for requirement in run_response.active_requirements:
2 if requirement.is_external_tool_execution:
3 print(f"Tool: {requirement.tool_execution.tool_name}")
4 print(f"Args: {requirement.tool_execution.tool_args}")
5
6 # Execute your custom logic here
7 result = my_custom_execution(requirement.tool_execution.tool_args)
8
9 # Set the result so the agent can continue
10 requirement.external_execution_result = result
11
12# After resolving the requirement, you can continue the run:
13response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)
Note

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 Kern will raise a ValueError, letting you know that not all requirements have been resolved.

Using Toolkits with External Execution

If you're using a Toolkit, you can specify which tools require external execution using the external_execution_required_tools parameter:

1from kern.tools.toolkit import Toolkit
2import subprocess
3
4class ShellTools(Toolkit):
5 def __init__(self, *args, **kwargs):
6 super().__init__(
7 tools=[self.list_dir, self.get_env],
8 external_execution_required_tools=["list_dir"], # Only this one needs external execution
9 *args,
10 **kwargs,
11 )
12
13 def list_dir(self, directory: str):
14 """Lists the contents of a directory."""
15 return subprocess.check_output(f"ls {directory}", shell=True).decode("utf-8")
16
17 def get_env(self, var_name: str):
18 """Gets an environment variable."""
19 import os
20 return os.getenv(var_name, "Not found")
21
22agent = Agent(
23 model=OpenAIResponses(id="gpt-5.2"),
24 tools=[ShellTools()],
25 markdown=True,
26)
27
28run_response = agent.run("What files are in my current directory and what's my PATH?")
29
30for requirement in run_response.active_requirements:
31 if requirement.is_external_tool_execution:
32 # Only list_dir will be here, get_env runs normally
33 if requirement.tool_execution.tool_name == "list_dir":
34 result = ShellTools().list_dir(**requirement.tool_execution.tool_args)
35 requirement.external_execution_result = result
36
37# After resolving the requirement, you can continue the run:
38response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)

This lets you mix external and internal tools in the same toolkit—perfect when you only need special handling for specific operations.

Mixed Tool Scenarios

You can absolutely have a mix of regular tools and external execution tools in the same agent.

When the agent wants to call multiple tools, only the ones marked with @tool(external_execution=True) will cause a pause:

1@tool(external_execution=True)
2def sensitive_database_query(query: str) -> str:
3 """Execute a database query."""
4 pass
5
6@tool
7def safe_calculation(x: int, y: int) -> int:
8 """Perform a safe calculation."""
9 return x + y
10
11agent = Agent(
12 model=OpenAIResponses(id="gpt-5.2"),
13 tools=[sensitive_database_query, safe_calculation],
14 markdown=True,
15)
16
17response = agent.run("Calculate 5 + 10 and query the users table")
18
19# Agent will pause when it tries to call sensitive_database_query
20# but safe_calculation executes normally
21
22for requirement in response.active_requirements:
23 if requirement.is_external_tool_execution:
24 if requirement.tool_execution.tool_name == "sensitive_database_query":
25 # Execute with your own DB connection and security checks
26 result = execute_safe_db_query(requirement.tool_execution.tool_args["query"])
27 requirement.external_execution_result = result
28
29# After resolving the requirement, you can continue the run:
30response = 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:

1import asyncio
2
3@tool(external_execution=True)
4async def async_external_tool(data: str) -> str:
5 """An async tool requiring external execution."""
6 pass
7
8agent = Agent(
9 model=OpenAIResponses(id="gpt-5.2"),
10 tools=[async_external_tool],
11 markdown=True,
12)
13
14async def main():
15 run_response = await agent.arun("Process some data")
16
17 for requirement in run_response.active_requirements:
18 if requirement.is_external_tool_execution:
19 # Execute your async external logic
20 result = await my_async_external_service(requirement.tool_execution.tool_args)
21 requirement.external_execution_result = result
22
23 response = await agent.acontinue_run(run_id=run_response.run_id, requirements=run_response.requirements)
24 print(response.content)
25
26asyncio.run(main())

Streaming Support

You can also use external execution with streaming responses:

1for run_event in agent.run("What files are in my directory?", stream=True):
2 if run_event.is_paused:
3 for requirement in run_event.active_requirements:
4 if requirement.is_external_tool_execution:
5 # Execute externally
6 result = execute_tool_externally(requirement.tool_execution.tool_args)
7 requirement.external_execution_result = result
8
9 # Continue streaming
10 for response in agent.continue_run(
11 run_id=run_event.run_id,
12 requirements=run_event.requirements,
13 stream=True
14 ):
15 print(response.content, end="")
16 else:
17 print(run_event.content, end="")

Best Practices

  1. Always set results: Make sure you set requirement.external_execution_result for all requirements before continuing
  2. Error handling: Wrap your external execution in try-catch blocks and provide meaningful error messages as results
  3. Security validation: Use external execution to add extra security checks before running sensitive operations
  4. Logging: Log all external executions for audit trails
  5. Timeouts: Consider adding timeouts to your external execution logic to prevent hanging
Warning

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