Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Generic Anthropic Messages API tool loop (client tools + server tools + pause_turn + programmatic tool calling).34STATUS: LIVE_VERIFIED 2026-09-18 — run with `.venv/bin/python examples/shared/tool-loop/anthropic_tool_loop.py`5(claude-haiku-4-5-20251001, two custom tools, one parallel round trip, ~1.5k tokens ≈ $0.002). Dependency-free:6requests go through scripts/live.py (keys from .env, every call logged to reports/live-requests.jsonl).78Contract implemented (see docs/tools/anthropic/tool-use-loop.md):9 * loop while stop_reason == "tool_use"; exit on end_turn / max_tokens / stop_sequence / refusal10 * stop_reason == "pause_turn" (server tools hit the iteration cap): re-send the assistant content as-is, same tools11 * run EVERY client tool_use of a turn (parallel calls), return one tool_result per id, all in ONE user message,12 tool_result blocks FIRST (text after them only when no server tool / programmatic call is pending)13 * a server_tool_use / mcp_tool_use with no matching *_tool_result in the same response is still pending:14 reply with tool_result blocks only, keep the same tools array (the API runs it on the next request)15 * tool_use.caller.type != "direct" => programmatic call from code_execution: reply must be tool_result-only and16 carry the response's container.id in the top-level `container` field17 * errors are returned to the model with is_error:true instead of raising18"""19from __future__ import annotations2021import json22import sys23from pathlib import Path24from typing import Any, Callable2526ROOT = Path(__file__).resolve().parents[3]27sys.path.insert(0, str(ROOT))28from scripts.live import anthropic_request # noqa: E4022930MODEL = "claude-haiku-4-5-20251001"31RESULT_SUFFIXES = ("_tool_result",) # web_search_tool_result, web_fetch_tool_result, code_execution_tool_result, mcp_tool_result, …323334def pending_server_calls(content: list[dict]) -> list[dict]:35 """server_tool_use / mcp_tool_use blocks whose result block is not in the same response."""36 answered = {b.get("tool_use_id") for b in content if b.get("type", "").endswith(RESULT_SUFFIXES)}37 return [b for b in content if b.get("type") in ("server_tool_use", "mcp_tool_use") and b["id"] not in answered]383940def run_tool_loop(messages: list[dict], tools: list[dict], handlers: dict[str, Callable[[dict], Any]], *,41 model: str = MODEL, max_tokens: int = 300, beta: str | None = None, max_turns: int = 8,42 extra: dict | None = None) -> dict:43 container: str | None = None44 last: dict = {}45 for turn in range(max_turns):46 body: dict = {"model": model, "max_tokens": max_tokens, "tools": tools, "messages": messages, **(extra or {})}47 if container:48 body["container"] = container49 status, resp, _ = anthropic_request("POST", "/v1/messages", body, beta=beta, note=f"anthropic_tool_loop turn {turn}")50 if status != 200:51 raise RuntimeError(f"HTTP {status}: {json.dumps(resp)[:400]}")52 last = resp53 if resp.get("container"):54 container = resp["container"]["id"] # keep code-execution state across turns55 content = resp["content"]56 messages.append({"role": "assistant", "content": content})57 stop = resp["stop_reason"]58 if stop == "pause_turn":59 continue # server-side loop paused: re-send as-is with the same tools60 if stop != "tool_use":61 return resp62 results: list[dict] = []63 programmatic = False64 for block in content:65 if block.get("type") != "tool_use":66 continue67 caller = (block.get("caller") or {}).get("type", "direct")68 programmatic |= caller != "direct"69 handler = handlers.get(block["name"])70 result: dict = {"type": "tool_result", "tool_use_id": block["id"]}71 if block.get("toolset_name"): # computer / browser toolset members must echo toolset_name72 result["toolset_name"] = block["toolset_name"]73 try:74 if handler is None:75 raise KeyError(f"no handler for tool {block['name']}")76 out = handler(block["input"])77 result["content"] = out if isinstance(out, (str, list)) else json.dumps(out)78 except Exception as err: # noqa: BLE001 — report to the model, do not crash the loop79 result["content"] = f"Error: {err}"80 result["is_error"] = True81 results.append(result)82 if not results: # only server tools pending and nothing for us: re-send as-is83 continue84 # Programmatic calls and pending server calls: the user message must contain tool_result blocks ONLY.85 if pending_server_calls(content) or programmatic:86 messages.append({"role": "user", "content": results})87 else:88 messages.append({"role": "user", "content": results}) # text could be appended AFTER results here89 return last909192if __name__ == "__main__":93 tools = [94 {"name": "get_weather", "description": "Get the current weather for a city.",95 "input_schema": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}},96 {"name": "get_time", "description": "Get the current local time in a city.",97 "input_schema": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}},98 ]99 handlers = {"get_weather": lambda i: f"{i['location']}: 18 C, light rain", "get_time": lambda i: f"{i['location']}: 14:05"}100 msgs = [{"role": "user", "content": "Weather and local time in Tokyo? Use both tools, then answer in one short sentence."}]101 final = run_tool_loop(msgs, tools, handlers, max_tokens=200)102 print(final["stop_reason"], json.dumps([b.get("text") for b in final["content"] if b["type"] == "text"]))103 print("turns:", sum(1 for m in msgs if m["role"] == "assistant"), "usage:", final["usage"]["input_tokens"], "in /", final["usage"]["output_tokens"], "out")104