External Tool Execution Stream Async

This example demonstrates how to execute tools outside of the agent using external tool execution with async streaming responses. It shows how to handle external tool execution in an asynchronous environment while maintaining real-time streaming.

Create a Python file

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

Set up your virtual environment

1uv venv --python 3.12
2source .venv/bin/activate

1uv venv --python 3.12
2.venv\Scripts\activate

Install dependencies

1uv pip install -U kern-ai openai

Export your OpenAI API key

1export OPENAI_API_KEY="your_openai_api_key_here"

1$Env:OPENAI_API_KEY="your_openai_api_key_here"

Run Agent

1python external_tool_execution_stream_async.py