Implementation:Openai Openai agents python LocalShellTool Pattern
| Knowledge Sources | |
|---|---|
| Domains | Tool_Integration, Shell_Execution, Local_Execution |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Demonstrates the LocalShellTool with a synchronous shell executor function that receives a LocalShellCommandRequest and returns command output as a string.
Description
This implementation shows how to use the LocalShellTool to give an agent the ability to execute shell commands on the local machine. The tool requires a user-defined executor function that receives a LocalShellCommandRequest object and returns the combined stdout and stderr as a string. The request object contains a data.action attribute with fields for command, working_directory, env, and timeout_ms.
The executor function in this example uses Python's subprocess.run() to execute commands synchronously. It supports configurable working directory (defaulting to the current directory), environment variable merging, timeout handling (converting milliseconds to seconds), and error capture. If a timeout occurs, a descriptive message is returned instead of raising an exception. Any other exceptions are caught and returned as error strings, ensuring the agent always receives a usable response.
The agent is configured with the codex-mini-latest model, which is noted as a compatible model for the local shell tool. This is a simpler alternative to the full ShellTool and does not include approval or human-in-the-loop mechanisms.
Usage
Use this pattern for building agents that need to execute local system commands in a straightforward, synchronous manner. This is suitable for development tools, system administration assistants, or build automation agents where the commands are expected to complete quickly and human approval is not required.
Code Reference
Source Location
- Repository: Openai_Openai_agents_python
- File: examples/tools/local_shell.py
- Lines: 1-45
Signature
def shell_executor(request: LocalShellCommandRequest) -> str:
args = request.data.action
completed = subprocess.run(
args.command,
cwd=args.working_directory or os.getcwd(),
env={**os.environ, **args.env} if args.env else os.environ,
capture_output=True,
text=True,
timeout=(args.timeout_ms / 1000) if args.timeout_ms else None,
)
return completed.stdout + completed.stderr
LocalShellTool(executor=shell_executor)
Import
from agents import Agent, LocalShellCommandRequest, LocalShellTool, Runner, trace
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| executor | Callable[[LocalShellCommandRequest], str] |
Yes | Synchronous function that executes the shell command and returns output as a string. |
| request.data.action.command | str |
Yes | The shell command string to execute. |
| request.data.action.working_directory | None | No | Working directory for command execution; defaults to current directory if not provided. |
| request.data.action.env | None | No | Additional environment variables to merge with the current environment. |
| request.data.action.timeout_ms | None | No | Command timeout in milliseconds; converted to seconds for subprocess.run().
|
Outputs
| Name | Type | Description |
|---|---|---|
| executor return | str |
Combined stdout and stderr from the executed command, or an error/timeout message. |
| result.final_output | str |
The agent's final text response incorporating the command output. |
Usage Examples
Basic Local Shell Agent
import asyncio
import os
import subprocess
from agents import Agent, LocalShellCommandRequest, LocalShellTool, Runner, trace
def shell_executor(request: LocalShellCommandRequest) -> str:
args = request.data.action
try:
completed = subprocess.run(
args.command,
cwd=args.working_directory or os.getcwd(),
env={**os.environ, **args.env} if args.env else os.environ,
capture_output=True,
text=True,
timeout=(args.timeout_ms / 1000) if args.timeout_ms else None,
)
return completed.stdout + completed.stderr
except subprocess.TimeoutExpired:
return "Command execution timed out"
except Exception as e:
return f"Error executing command: {str(e)}"
async def main():
agent = Agent(
name="Shell Assistant",
instructions="You are a helpful assistant that can execute shell commands.",
model="codex-mini-latest",
tools=[LocalShellTool(executor=shell_executor)],
)
with trace("Local shell example"):
result = await Runner.run(
agent,
"List the files in the current directory and tell me how many there are.",
)
print(result.final_output)
asyncio.run(main())