External Tool Execution Async

1import asyncio
2import subprocess
3
4from kern.agent import Agent
5from kern.db.sqlite import SqliteDb
6from kern.models.openai import OpenAIResponses
7from kern.tools import tool
8from kern.utils import pprint
9
10
11# We have to create a tool with the correct name, arguments and docstring for the agent to know what to call.
12@tool(external_execution=True)
13def execute_shell_command(command: str) -> str:
14    """Execute a shell command.
15
16    Args:
17        command (str): The shell command to execute
18
19    Returns:
20        str: The output of the shell command
21    """
22    if command.startswith("ls"):
23        return subprocess.check_output(command, shell=True).decode("utf-8")
24    else:
25        raise Exception(f"Unsupported command: {command}")
26
27
28agent = Agent(
29    model=OpenAIResponses(id="gpt-5.2"),
30    tools=[execute_shell_command],
31    markdown=True,
32    db=SqliteDb(session_table="test_session", db_file="tmp/example.db"),
33)
34
35run_response = asyncio.run(agent.arun("What files do I have in my current directory?"))
36if run_response.is_paused:
37    for requirement in run_response.active_requirements:
38        if requirement.needs_external_execution:
39            if requirement.tool_execution.tool_name == execute_shell_command.name:
40                print(
41                    f"Executing {requirement.tool_execution.tool_name} with args {requirement.tool_execution.tool_args} externally"
42                )
43                # We execute the tool ourselves. You can also execute something completely external here.
44                result = execute_shell_command.entrypoint(
45                    **requirement.tool_execution.tool_args
46                )  # type: ignore
47                # We have to set the result on the tool execution object so that the agent can continue
48                requirement.set_external_execution_result(result)
49
50run_response = asyncio.run(
51    agent.acontinue_run(
52        run_id=run_response.run_id,
53        requirements=run_response.requirements,
54    )
55)
56pprint.pprint_run_response(run_response)
57
58
59# Or for simple debug flow
60# agent.print_response("What files do I have in my current directory?")

1uv pip install -U kern-ai openai

1python external_tool_execution_async.py