Add new tools for gpt-5.1

examples/tools/apply_patch.py

@@ -0,0 +1,169 @@
+import argparse
+import asyncio
+import hashlib
+import os
+import tempfile
+from pathlib import Path
+
+from agents import Agent, ApplyPatchTool, ModelSettings, Runner, apply_diff, trace
+from agents.editor import ApplyPatchOperation, ApplyPatchResult
+
+
+class ApprovalTracker:
+    def __init__(self) -> None:
+        self._approved: set[str] = set()
+
+    def fingerprint(self, operation: ApplyPatchOperation, relative_path: str) -> str:
+        hasher = hashlib.sha256()
+        hasher.update(operation.type.encode("utf-8"))
+        hasher.update(b"\0")
+        hasher.update(relative_path.encode("utf-8"))
+        hasher.update(b"\0")
+        hasher.update((operation.diff or "").encode("utf-8"))
+        return hasher.hexdigest()
+
+    def remember(self, fingerprint: str) -> None:
+        self._approved.add(fingerprint)
+
+    def is_approved(self, fingerprint: str) -> bool:
+        return fingerprint in self._approved
+
+
+class WorkspaceEditor:
+    def __init__(self, root: Path, approvals: ApprovalTracker, auto_approve: bool) -> None:
+        self._root = root.resolve()
+        self._approvals = approvals
+        self._auto_approve = auto_approve or os.environ.get("APPLY_PATCH_AUTO_APPROVE") == "1"
+
+    def create_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult:
+        relative = self._relative_path(operation.path)
+        self._require_approval(operation, relative)
+        target = self._resolve(operation.path, ensure_parent=True)
+        diff = operation.diff or ""
+        content = apply_diff("", diff, mode="create")
+        target.write_text(content, encoding="utf-8")
+        return ApplyPatchResult(output=f"Created {relative}")
+
+    def update_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult:
+        relative = self._relative_path(operation.path)
+        self._require_approval(operation, relative)
+        target = self._resolve(operation.path)
+        original = target.read_text(encoding="utf-8")
+        diff = operation.diff or ""
+        patched = apply_diff(original, diff)
+        target.write_text(patched, encoding="utf-8")
+        return ApplyPatchResult(output=f"Updated {relative}")
+
+    def delete_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult:
+        relative = self._relative_path(operation.path)
+        self._require_approval(operation, relative)
+        target = self._resolve(operation.path)
+        target.unlink(missing_ok=True)
+        return ApplyPatchResult(output=f"Deleted {relative}")
+
+    def _relative_path(self, value: str) -> str:
+        resolved = self._resolve(value)
+        return resolved.relative_to(self._root).as_posix()
+
+    def _resolve(self, relative: str, ensure_parent: bool = False) -> Path:
+        candidate = Path(relative)
+        target = candidate if candidate.is_absolute() else (self._root / candidate)
+        target = target.resolve()
+        try:
+            target.relative_to(self._root)
+        except ValueError:
+            raise RuntimeError(f"Operation outside workspace: {relative}") from None
+        if ensure_parent:
+            target.parent.mkdir(parents=True, exist_ok=True)
+        return target
+
+    def _require_approval(self, operation: ApplyPatchOperation, display_path: str) -> None:
+        fingerprint = self._approvals.fingerprint(operation, display_path)
+        if self._auto_approve or self._approvals.is_approved(fingerprint):
+            self._approvals.remember(fingerprint)
+            return
+
+        print("\n[apply_patch] approval required")
+        print(f"- type: {operation.type}")
+        print(f"- path: {display_path}")
+        if operation.diff:
+            preview = operation.diff if len(operation.diff) < 400 else f"{operation.diff[:400]}…"
+            print("- diff preview:\n", preview)
+        answer = input("Proceed? [y/N] ").strip().lower()
+        if answer not in {"y", "yes"}:
+            raise RuntimeError("Apply patch operation rejected by user.")
+        self._approvals.remember(fingerprint)
+
+
+async def main(auto_approve: bool, model: str) -> None:
+    with trace("apply_patch_example"):
+        with tempfile.TemporaryDirectory(prefix="apply-patch-example-") as workspace:
+            workspace_path = Path(workspace).resolve()
+            approvals = ApprovalTracker()
+            editor = WorkspaceEditor(workspace_path, approvals, auto_approve)
+            tool = ApplyPatchTool(editor=editor)
+            previous_response_id: str | None = None
+
+            agent = Agent(
+                name="Patch Assistant",
+                model=model,
+                instructions=(
+                    f"You can edit files inside {workspace_path} using the apply_patch tool. "
+                    "When modifying an existing file, include the file contents between "
+                    "<BEGIN_FILES> and <END_FILES> in your prompt."
+                ),
+                tools=[tool],
+                model_settings=ModelSettings(tool_choice="required"),
+            )
+
+            print(f"[info] Workspace root: {workspace_path}")
+            print(f"[info] Using model: {model}")
+            print("[run] Creating tasks.md")
+            result = await Runner.run(
+                agent,
+                "Create tasks.md with a shopping checklist of 5 entries.",
+                previous_response_id=previous_response_id,
+            )
+            previous_response_id = result.last_response_id
+            print(f"[run] Final response #1:\n{result.final_output}\n")
+            notes_path = workspace_path / "tasks.md"
+            if not notes_path.exists():
+                raise RuntimeError(f"{notes_path} was not created by the apply_patch tool.")
+            updated_notes = notes_path.read_text(encoding="utf-8")
+            print("[file] tasks.md after creation:\n")
+            print(updated_notes)
+
+            prompt = (
+                "<BEGIN_FILES>\n"
+                f"===== tasks.md\n{updated_notes}\n"
+                "<END_FILES>\n"
+                "Check off the last two items from the file."
+            )
+            print("\n[run] Updating tasks.md")
+            result2 = await Runner.run(
+                agent,
+                prompt,
+                previous_response_id=previous_response_id,
+            )
+            print(f"[run] Final response #2:\n{result2.final_output}\n")
+            if not notes_path.exists():
+                raise RuntimeError("tasks.md vanished unexpectedly before the second read.")
+            print("[file] Final tasks.md:\n")
+            print(notes_path.read_text(encoding="utf-8"))
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--auto-approve",
+        action="store_true",
+        default=False,
+        help="Skip manual confirmations for apply_patch operations.",
+    )
+    parser.add_argument(
+        "--model",
+        default="gpt-5.1",
+        help="Model ID to use for the agent.",
+    )
+    args = parser.parse_args()
+    asyncio.run(main(args.auto_approve, args.model))

examples/tools/code_interpreter.py

@@ -1,8 +1,16 @@
 import asyncio
+from collections.abc import Mapping
+from typing import Any
 
 from agents import Agent, CodeInterpreterTool, Runner, trace
 
 
+def _get_field(obj: Any, key: str) -> Any:
+    if isinstance(obj, Mapping):
+        return obj.get(key)
+    return getattr(obj, key, None)
+
+
 async def main():
     agent = Agent(
         name="Code interpreter",
@@ -21,14 +29,19 @@ async def main():
         print("Solving math problem...")
         result = Runner.run_streamed(agent, "What is the square root of273 * 312821 plus 1782?")
         async for event in result.stream_events():
-            if (
-                event.type == "run_item_stream_event"
-                and event.item.type == "tool_call_item"
-                and event.item.raw_item.type == "code_interpreter_call"
-            ):
-                print(f"Code interpreter code:\n```\n{event.item.raw_item.code}\n```\n")
-            elif event.type == "run_item_stream_event":
-                print(f"Other event: {event.item.type}")
+            if event.type != "run_item_stream_event":
+                continue
+
+            item = event.item
+            if item.type == "tool_call_item":
+                raw_call = item.raw_item
+                if _get_field(raw_call, "type") == "code_interpreter_call":
+                    code = _get_field(raw_call, "code")
+                    if isinstance(code, str):
+                        print(f"Code interpreter code:\n```\n{code}\n```\n")
+                        continue
+
+            print(f"Other event: {event.item.type}")
 
         print(f"Final output: {result.final_output}")
 

examples/tools/image_generator.py

@@ -4,10 +4,18 @@
 import subprocess
 import sys
 import tempfile
+from collections.abc import Mapping
+from typing import Any
 
 from agents import Agent, ImageGenerationTool, Runner, trace
 
 
+def _get_field(obj: Any, key: str) -> Any:
+    if isinstance(obj, Mapping):
+        return obj.get(key)
+    return getattr(obj, key, None)
+
+
 def open_file(path: str) -> None:
     if sys.platform.startswith("darwin"):
         subprocess.run(["open", path], check=False)  # macOS
@@ -37,17 +45,23 @@ async def main():
         )
         print(result.final_output)
         for item in result.new_items:
-            if (
-                item.type == "tool_call_item"
-                and item.raw_item.type == "image_generation_call"
-                and (img_result := item.raw_item.result)
-            ):
-                with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
-                    tmp.write(base64.b64decode(img_result))
-                    temp_path = tmp.name
-
-                # Open the image
-                open_file(temp_path)
+            if item.type != "tool_call_item":
+                continue
+
+            raw_call = item.raw_item
+            call_type = _get_field(raw_call, "type")
+            if call_type != "image_generation_call":
+                continue
+
+            img_result = _get_field(raw_call, "result")
+            if not isinstance(img_result, str):
+                continue
+
+            with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
+                tmp.write(base64.b64decode(img_result))
+                temp_path = tmp.name
+
+            open_file(temp_path)
 
 
 if __name__ == "__main__":

examples/tools/shell.py

@@ -0,0 +1,114 @@
+import argparse
+import asyncio
+import os
+from collections.abc import Sequence
+from pathlib import Path
+
+from agents import (
+    Agent,
+    ModelSettings,
+    Runner,
+    ShellCallOutcome,
+    ShellCommandOutput,
+    ShellCommandRequest,
+    ShellResult,
+    ShellTool,
+    trace,
+)
+
+
+class ShellExecutor:
+    """Executes shell commands with optional approval."""
+
+    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
+        await require_approval(action.commands)
+
+        outputs: list[ShellCommandOutput] = []
+        for command in action.commands:
+            proc = await asyncio.create_subprocess_shell(
+                command,
+                cwd=self.cwd,
+                env=os.environ.copy(),
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            timed_out = False
+            try:
+                timeout = (action.timeout_ms or 0) / 1000 or None
+                stdout_bytes, stderr_bytes = await asyncio.wait_for(
+                    proc.communicate(), timeout=timeout
+                )
+            except asyncio.TimeoutError:
+                proc.kill()
+                stdout_bytes, stderr_bytes = await proc.communicate()
+                timed_out = True
+
+            stdout = stdout_bytes.decode("utf-8", errors="ignore")
+            stderr = stderr_bytes.decode("utf-8", errors="ignore")
+            outputs.append(
+                ShellCommandOutput(
+                    command=command,
+                    stdout=stdout,
+                    stderr=stderr,
+                    outcome=ShellCallOutcome(
+                        type="timeout" if timed_out else "exit",
+                        exit_code=getattr(proc, "returncode", None),
+                    ),
+                )
+            )
+
+            if timed_out:
+                break
+
+        return ShellResult(
+            output=outputs,
+            provider_data={"working_directory": str(self.cwd)},
+        )
+
+
+async def require_approval(commands: Sequence[str]) -> None:
+    if os.environ.get("SHELL_AUTO_APPROVE") == "1":
+        return
+    print("Shell command approval required:")
+    for entry in commands:
+        print(" ", entry)
+    response = input("Proceed? [y/N] ").strip().lower()
+    if response not in {"y", "yes"}:
+        raise RuntimeError("Shell command execution rejected by user.")
+
+
+async def main(prompt: str, model: str) -> None:
+    with trace("shell_example"):
+        print(f"[info] Using model: {model}")
+        agent = Agent(
+            name="Shell Assistant",
+            model=model,
+            instructions=(
+                "You can run shell commands using the shell tool. "
+                "Keep responses concise and include command output when helpful."
+            ),
+            tools=[ShellTool(executor=ShellExecutor())],
+            model_settings=ModelSettings(tool_choice="required"),
+        )
+
+        result = await Runner.run(agent, prompt)
+        print(f"\nFinal response:\n{result.final_output}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--prompt",
+        default="Show the list of files in the current directory.",
+        help="Instruction to send to the agent.",
+    )
+    parser.add_argument(
+        "--model",
+        default="gpt-5.1",
+    )
+    args = parser.parse_args()
+    asyncio.run(main(args.prompt, args.model))