ReAct Loop

Python. Use this when you want the loop written by hand, with budget and errors handled in plain sight.

import subprocess

import anthropic

client = anthropic.Anthropic()
MAX_STEPS = 20

TOOLS = [
    {
        "name": "read_file",
        "description": "Reads a repository file. Use it to view test or production code.",
        "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]},
    },
    {
        "name": "run_tests",
        "description": "Runs pytest on a target (file or file::test) and returns the output.",
        "input_schema": {"type": "object", "properties": {"target": {"type": "string"}}, "required": ["target"]},
    },
]


def execute(name: str, args: dict) -> tuple[str, bool]:
    try:
        if name == "read_file":
            return open(args["path"], encoding="utf-8").read()[:20_000], False
        if name == "run_tests":
            out = subprocess.run(["pytest", args["target"], "-x", "-q"], capture_output=True, text=True, timeout=120)
            return (out.stdout + out.stderr)[-8_000:], False
        return f"unknown tool: {name}", True
    except Exception as e:  # 6. the error comes back as an observation
        return f"{type(e).__name__}: {e}", True


def investigate(task: str) -> str:
    messages = [{"role": "user", "content": task}]
    for step in range(MAX_STEPS):  # 4. step budget
        reply = client.messages.create(
            model="claude-opus-5-5", max_tokens=8000, tools=TOOLS, messages=messages,
            system="You investigate failing tests. Answer with the cause and the evidence.",
        )
        if reply.stop_reason != "tool_use":  # 8. final answer
            return next(b.text for b in reply.content if b.type == "text")

        messages.append({"role": "assistant", "content": reply.content})
        results = []
        for block in reply.content:
            if block.type == "tool_use":  # 5. execute
                content, is_error = execute(block.name, block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": content, "is_error": is_error})
        messages.append({"role": "user", "content": results})

    return "Stopped at the step limit without a conclusion. See the trace."  # 7

What you gained: every turn, every error, and the cap are visible. What you lost: tracing, retries, and streaming are on you.

TypeScript. Use this when the loop runs on Node and you prefer the SDK to handle the turns.

import Anthropic from "@anthropic-ai/sdk";
import { betaZodTool } from "@anthropic-ai/sdk/helpers/beta/zod";
import { execFile } from "node:child_process";
import { readFile } from "node:fs/promises";
import { promisify } from "node:util";
import { z } from "zod";

const client = new Anthropic();
const run = promisify(execFile);

const readFileTool = betaZodTool({
  name: "read_file",
  description: "Reads a repository file. Use it to view test or production code.",
  inputSchema: z.object({ path: z.string() }),
  run: async ({ path }) => (await readFile(path, "utf8")).slice(0, 20_000),
});

const runTests = betaZodTool({
  name: "run_tests",
  description: "Runs a file's tests with vitest and returns the output.",
  inputSchema: z.object({ target: z.string() }),
  run: async ({ target }) => {
    try {
      const { stdout } = await run("npx", ["vitest", "run", target], { timeout: 120_000 });
      return stdout.slice(-8_000);
    } catch (e: any) {
      return String(e.stdout ?? e.message).slice(-8_000); // 6. a failing test is an observation too
    }
  },
});

export async function investigate(task: string): Promise<string> {
  // 2 to 8. the runner repeats model and tools until the final answer
  const runner = client.beta.messages.toolRunner({
    model: "claude-opus-5-5",
    max_tokens: 8000,
    system: "You investigate failing tests. Answer with the cause and the evidence.",
    tools: [readFileTool, runTests],
    messages: [{ role: "user", content: task }],
    max_iterations: 20, // 4. turn cap
  });
  const final = await runner;
  return final.content.find((b) => b.type === "text")?.text ?? "";
}

What you gained: the loop comes ready and arguments arrive validated. What you lost: fine control of each turn; for a dollar budget, use the runner’s hooks or go back to the manual loop.

Python + LangChain. Use this when you already understand the hand-written loop and want the same investigator with a provider swap.

import subprocess

from langchain.agents import create_agent
from langchain.tools import tool


@tool
def read_file(path: str) -> str:
    """Read a file in the repository. Use it to see test code or production code."""
    try:
        return open(path, encoding="utf-8").read()[:20_000]
    except OSError as exc:
        return f"error: {exc}"  # 6. the error comes back as an observation


@tool
def run_tests(target: str) -> str:
    """Run pytest on a target (a file or file::test) and return the output."""
    try:
        out = subprocess.run(["pytest", target, "-x", "-q"], capture_output=True, text=True, timeout=120)
        return (out.stdout + out.stderr)[-8_000:]
    except (OSError, subprocess.TimeoutExpired) as exc:
        return f"error: {exc}"


agent = create_agent(
    "openai:gpt-4o",  # or "anthropic:claude-opus-5-5"
    tools=[read_file, run_tests],
    system_prompt="You investigate failing tests. Answer with the cause and the evidence.",
)


def investigate(task: str) -> str:
    # 2 to 8. reasoning, tool, and observation until the answer or the cap
    result = agent.invoke(
        {"messages": [{"role": "user", "content": task}]},
        {"recursion_limit": 40},  # 4. cap; each tool counts as a step
    )
    return result["messages"][-1].text

What you gained: the ReAct loop without writing the while, and the model is a string. What you lost: the money stop and the “I could not finish” message when the cap hits. Those stay on the Python tab.

Python + LangGraph. Use this when you want the loop as nodes: model, tools, and the edge that decides to stop.

from typing import Annotated, Literal

from langchain.chat_models import init_chat_model
from langchain_core.messages import ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int

model = init_chat_model("openai:gpt-4o", temperature=0).bind_tools([read_file, run_tests])

def call_model(state: State) -> dict:
    # 2. the model picks the next step
    if state["steps"] >= 20:
        return {"messages": [("assistant", "Stopped: step cap.")]}  # 7
    reply = model.invoke(state["messages"])
    return {"messages": [reply], "steps": state["steps"] + 1}

def run_tools(state: State) -> dict:
    # 5 and 6. run the tool and return the observation, including the error
    notes = []
    for call in state["messages"][-1].tool_calls:
        try:
            text = read_file.invoke(call["args"]) if call["name"] == "read_file" else run_tests.invoke(call["args"])
        except Exception as exc:
            text = f"error: {exc}"
        notes.append(ToolMessage(content=str(text), tool_call_id=call["id"]))
    return {"messages": notes}

def route(state: State) -> Literal["tools", "end"]:
    # 3. tool or answer, from tool_calls, not from the prose
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else "end"

graph = StateGraph(State)
graph.add_node("model", call_model)
graph.add_node("tools", run_tools)
graph.add_edge(START, "model")
graph.add_conditional_edges("model", route, {"tools": "tools", "end": END})
graph.add_edge("tools", "model")
app = graph.compile()

read_file and run_tests are the @tool functions from the Python + LangChain tab. bind_tools attaches the schema to the model; the tools node is your code.

What you gained: you can insert a node of yours between the model and the tool. What you lost: more code than create_agent, which is this graph already assembled.

Python + OpenAI Agents. Use this when the loop is OpenAI's and the cap is `max_turns`.

from agents import Agent, Runner, function_tool

@function_tool
def read_file(path: str) -> str:
    """Read a file in the repository. Use it to see test code or production code."""
    try:
        return open(path, encoding="utf-8").read()[:20_000]
    except OSError as exc:
        return f"error: {exc}"

@function_tool
def run_tests(target: str) -> str:
    """Run pytest on a target and return the output."""
    try:
        out = subprocess.run(["pytest", target, "-x", "-q"], capture_output=True, text=True, timeout=120)
        return (out.stdout + out.stderr)[-8_000:]
    except (OSError, subprocess.TimeoutExpired) as exc:
        return f"error: {exc}"

agent = Agent(
    name="investigator",
    instructions="You investigate failing tests. Answer with the cause and the evidence.",
    tools=[read_file, run_tests],
    model="gpt-4o",
)

def investigate(task: str) -> str:
    # 4. max_turns is this runner's turn cap
    result = Runner.run_sync(agent, task, max_turns=20)
    return str(result.final_output)

What you gained: the loop and the turn cap in the OpenAI SDK. What you lost: money and the stop message stay yours; max_turns only counts model calls.

Python + Pydantic AI. Use this when the tool argument has to be a Python type and the model is a provider string.

from pydantic_ai import Agent
from pydantic_ai.usage import UsageLimits

agent = Agent(
    "openai:gpt-4o",  # or "anthropic:claude-opus-5-5"
    system_prompt="You investigate failing tests. Answer with the cause and the evidence.",
)

@agent.tool_plain
def read_file(path: str) -> str:
    """Read a file in the repository."""
    try:
        return open(path, encoding="utf-8").read()[:20_000]
    except OSError as exc:
        return f"error: {exc}"

@agent.tool_plain
def run_tests(target: str) -> str:
    """Run pytest on a target and return the output."""
    try:
        out = subprocess.run(["pytest", target, "-x", "-q"], capture_output=True, text=True, timeout=120)
        return (out.stdout + out.stderr)[-8_000:]
    except (OSError, subprocess.TimeoutExpired) as exc:
        return f"error: {exc}"

def investigate(task: str) -> str:
    # request_limit is the cap on model calls
    return agent.run_sync(task, usage_limits=UsageLimits(request_limit=20)).output

What you gained: the tool schema comes from the Python signature. What you lost: the same incomplete cap; time and money are not in UsageLimits.

TypeScript + AI SDK. Use this when the agent lives in a web app and the model changes without rewriting the loop.

import { readFile } from "node:fs/promises";
import { generateText, stepCountIs, tool } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";

export async function investigate(task: string): Promise<string> {
  const result = await generateText({
    model: anthropic("claude-opus-5-5"), // or openai("gpt-4o")
    system: "You investigate failing tests. Answer with the cause and the evidence.",
    prompt: task,
    stopWhen: stepCountIs(20), // 4. step cap
    tools: {
      read_file: tool({
        description: "Read a file in the repository.",
        inputSchema: z.object({ path: z.string() }),
        execute: async ({ path }) => {
          try {
            return await readFile(path, "utf8");
          } catch (err) {
            return `error: ${err}`; // 6
          }
        },
      }),
      run_tests: tool({
        description: "Run pytest on a target and return the output.",
        inputSchema: z.object({ target: z.string() }),
        execute: async ({ target }) => runPytest(target),
      }),
    },
  });
  return result.text; // 8
}

runPytest is yours. Swap anthropic(...) for openai(...) without touching the tools.

What you gained: streaming and UI in the same SDK, with the loop included. What you lost: the cap is steps, not money.