Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Generic OpenAI Responses API tool loop (function calls, custom tools, MCP approvals, shell/apply_patch stubs).34STATUS: LIVE_VERIFIED 2026-09-18 — run with `.venv/bin/python examples/shared/tool-loop/openai_tool_loop.py` (gpt-5.4-nano,5one get_weather round trip, ~70 tokens). Keys come from .env via scripts/live.py; every request is logged to6reports/live-requests.jsonl.78Patterns implemented (see docs/openai/tool-loop.md):9 * loop until the response contains no actionable items (function_call, custom_tool_call, mcp_approval_request,10 shell_call with local environment, apply_patch_call)11 * execute ALL calls of a turn before sending outputs back (parallel_tool_calls default true)12 * chain with previous_response_id (server-side state) — reasoning items are then carried automatically13 * bounded number of turns, error strings returned to the model instead of raising14"""15from __future__ import annotations1617import json18import sys19from pathlib import Path20from typing import Any, Callable2122ROOT = Path(__file__).resolve().parents[3]23sys.path.insert(0, str(ROOT))24from scripts.live import load_env, log_request # noqa: E4022526load_env()27from openai import OpenAI # noqa: E402 (installed SDK)2829client = OpenAI()30MODEL = "gpt-5.4-nano"3132ToolFn = Callable[[dict[str, Any]], Any]333435def run_tool_loop(input_text: str, tools: list[dict], handlers: dict[str, ToolFn], *, model: str = MODEL,36 max_turns: int = 6, approve_mcp: Callable[[dict], bool] | None = None, verbose: bool = True) -> str:37 """Returns the final assistant text. `handlers` maps function/custom tool names to callables."""38 kwargs: dict[str, Any] = {"model": model, "tools": tools, "input": input_text, "max_output_tokens": 256,39 "reasoning": {"effort": "low"}}40 for turn in range(max_turns):41 resp = client.responses.create(**kwargs)42 log_request("openai", "POST", "/v1/responses", 200, 0.0, f"tool loop turn {turn} ({model})")43 outputs: list[dict] = []44 for item in resp.output:45 t = item.type46 if t == "function_call":47 fn = handlers.get(item.name)48 try:49 result = fn(json.loads(item.arguments or "{}")) if fn else {"error": f"unknown tool {item.name}"}50 except Exception as e: # noqa: BLE001 — surface errors to the model, never crash the loop51 result = {"error": str(e)[:500]}52 outputs.append({"type": "function_call_output", "call_id": item.call_id, "output": json.dumps(result)})53 elif t == "custom_tool_call":54 fn = handlers.get(item.name)55 result = fn({"input": item.input}) if fn else {"error": "unknown custom tool"}56 outputs.append({"type": "custom_tool_call_output", "call_id": item.call_id, "output": json.dumps(result)})57 elif t == "mcp_approval_request":58 ok = bool(approve_mcp and approve_mcp(item.model_dump()))59 outputs.append({"type": "mcp_approval_response", "approval_request_id": item.id, "approve": ok,60 **({} if ok else {"reason": "denied by policy"})})61 elif t == "shell_call" and (item.environment is None or getattr(item.environment, "type", None) == "local"):62 # Local shell: YOU execute. This stub refuses everything except a trivial allow-list.63 cmds = item.action.commands64 allowed = all(c.strip() in ("echo OK", "pwd") for c in cmds)65 out = [{"stdout": "OK\n" if c.strip() == "echo OK" else "", "stderr": "" if allowed else "blocked by policy",66 "outcome": {"type": "exit", "exit_code": 0 if allowed else 126}} for c in cmds]67 outputs.append({"type": "shell_call_output", "call_id": item.call_id, "output": out})68 elif t == "apply_patch_call":69 # Never apply blindly: validate path, then report status completed|failed.70 op = item.operation71 bad = ".." in op.path or op.path.startswith("/")72 outputs.append({"type": "apply_patch_call_output", "call_id": item.call_id,73 "status": "failed" if bad else "completed",74 "output": "path rejected" if bad else f"{op.type} {op.path} applied (dry run)"})75 if verbose:76 print(f"turn {turn}: output types = {[i.type for i in resp.output]}")77 if not outputs:78 return resp.output_text79 kwargs = {"model": model, "tools": tools, "previous_response_id": resp.id, "input": outputs,80 "max_output_tokens": 256, "reasoning": {"effort": "low"}}81 raise RuntimeError("tool loop did not converge")828384if __name__ == "__main__":85 weather_tool = {"type": "function", "name": "get_weather", "description": "Current weather for a city.",86 "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"],87 "additionalProperties": False}, "strict": True}88 text = run_tool_loop("What's the weather in Paris? Answer in 5 words.", [weather_tool],89 {"get_weather": lambda a: {"city": a["city"], "temp_c": 18, "sky": "clear"}})90 print("final:", text)91