Implementation:Openai Openai agents python ShellTool Pattern
| Knowledge Sources | |
|---|---|
| Domains | Tool_Integration, Shell_Execution, Human_In_The_Loop |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Demonstrates the ShellTool with an async executor class, multi-command execution with timeout support, and a human-in-the-loop approval callback for gating command execution.
Description
This implementation provides a full-featured example of the ShellTool, which is the more advanced shell integration compared to LocalShellTool. The core executor is the ShellExecutor class, an async callable that processes a ShellCommandRequest containing one or more commands. Each command is executed via asyncio.create_subprocess_shell() with configurable timeout support. The executor returns a ShellResult containing a list of ShellCommandOutput entries, each with stdout, stderr, and a ShellCallOutcome indicating whether the command exited normally or timed out.
The key differentiator of this implementation is the human-in-the-loop approval mechanism. When needs_approval=True is set on the ShellTool, the SDK invokes the on_approval callback before executing commands. The callback receives a RunContextWrapper and a ToolApprovalItem, from which the pending commands can be extracted. It returns a ShellOnApprovalFunctionResult dictionary with an "approve" boolean and an optional "reason" string. The example implements a CLI prompt that displays the commands and waits for user confirmation, with an auto-approve mode controlled by the SHELL_AUTO_APPROVE environment variable.
The executor maintains a configurable working directory (cwd) and processes commands sequentially, stopping early if a timeout occurs. The agent is configured with ModelSettings(tool_choice="required") to ensure the model always invokes the shell tool.
Usage
Use this pattern when building agents that execute shell commands in environments where security and auditability are important. The approval callback enables human oversight of potentially dangerous commands before they run, making this suitable for production environments, CI/CD agents, or interactive development assistants where untrusted prompts might lead to harmful commands.
Code Reference
Source Location
- Repository: Openai_Openai_agents_python
- File: examples/tools/shell.py
- Lines: 1-141
Signature
class ShellExecutor:
def __init__(self, cwd: Path | None = None):
self.cwd = Path(cwd or Path.cwd())
async def __call__(self, request: ShellCommandRequest) -> ShellResult:
...
ShellTool(
executor=ShellExecutor(),
needs_approval=True,
on_approval=on_shell_approval,
)
Import
from agents import (
Agent,
ModelSettings,
Runner,
ShellCallOutcome,
ShellCommandOutput,
ShellCommandRequest,
ShellResult,
ShellTool,
trace,
)
from agents.items import ToolApprovalItem
from agents.run_context import RunContextWrapper
from agents.tool import ShellOnApprovalFunctionResult
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| executor | ShellExecutor (async callable) |
Yes | Async callable that receives a ShellCommandRequest and returns a ShellResult.
|
| needs_approval | bool |
No | When True, triggers the on_approval callback before command execution.
|
| on_approval | Callable[[RunContextWrapper, ToolApprovalItem], Awaitable[ShellOnApprovalFunctionResult]] |
No | Async callback invoked for human approval; returns {"approve": bool, "reason": str}.
|
| request.data.action.commands | list[str] |
Yes | List of shell command strings to execute sequentially. |
| request.data.action.timeout_ms | None | No | Timeout in milliseconds for each command; None means no timeout.
|
| cwd | None | No | Working directory for command execution; defaults to current working directory. |
Outputs
| Name | Type | Description |
|---|---|---|
| ShellResult.output | list[ShellCommandOutput] |
List of output entries, one per executed command. |
| ShellCommandOutput.command | str |
The command that was executed. |
| ShellCommandOutput.stdout | str |
Standard output from the command. |
| ShellCommandOutput.stderr | str |
Standard error from the command. |
| ShellCommandOutput.outcome | ShellCallOutcome |
Outcome with type ("exit" or "timeout") and optional exit_code.
|
| ShellResult.provider_data | dict |
Additional metadata, e.g., {"working_directory": "/path"}.
|
| result.final_output | str |
The agent's final text response incorporating the command results. |
Usage Examples
Shell Tool with Human Approval
import asyncio
from pathlib import Path
from agents import (
Agent,
ModelSettings,
Runner,
ShellCallOutcome,
ShellCommandOutput,
ShellCommandRequest,
ShellResult,
ShellTool,
trace,
)
from agents.items import ToolApprovalItem
from agents.run_context import RunContextWrapper
from agents.tool import ShellOnApprovalFunctionResult
class ShellExecutor:
def __init__(self, cwd: Path | None = None):
self.cwd = Path(cwd or Path.cwd())
async def __call__(self, request: ShellCommandRequest) -> ShellResult:
action = request.data.action
outputs: list[ShellCommandOutput] = []
for command in action.commands:
proc = await asyncio.create_subprocess_shell(
command,
cwd=self.cwd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout_bytes, stderr_bytes = await proc.communicate()
outputs.append(
ShellCommandOutput(
command=command,
stdout=stdout_bytes.decode(),
stderr=stderr_bytes.decode(),
outcome=ShellCallOutcome(
type="exit",
exit_code=proc.returncode,
),
)
)
return ShellResult(
output=outputs,
provider_data={"working_directory": str(self.cwd)},
)
async def on_shell_approval(
_context: RunContextWrapper, approval_item: ToolApprovalItem
) -> ShellOnApprovalFunctionResult:
# Auto-approve for this example
return {"approve": True, "reason": "approved"}
async def main():
with trace("shell_example"):
agent = Agent(
name="Shell Assistant",
model="gpt-5.2",
instructions="You can run shell commands using the shell tool.",
tools=[
ShellTool(
executor=ShellExecutor(),
needs_approval=True,
on_approval=on_shell_approval,
)
],
model_settings=ModelSettings(tool_choice="required"),
)
result = await Runner.run(agent, "Show the list of files in the current directory.")
print(result.final_output)
asyncio.run(main())