LangGraph: Build an Agent with Tool Calling and Bounded Execution for Code Verification

Autonomous code modification requires more than a standard linear prompt chain. When an agent modifies a repository, it must read files, apply edits, execute test suites, inspect outputs, and self-correct when regressions appear. However, cyclical agent patterns—often based on the ReAct (Reasoning and Acting) paradigm—pose operational hazards if unconstrained. An agent encountering broken dependencies or persistent test failures can enter expensive, unbounded execution loops.

Using Google AI's documented ReAct architecture for LangGraph, this guide adapts core state-graph mechanics into a specialized, bounded coding agent. The implementation executes code edits, verifies changes against automated test suites, and enforces strict iteration budgets.

---

Core Architecture: State, Nodes, and Conditional Edges

LangGraph models agent workflows as directed graphs. Unlike static directed acyclic graphs (DAGs), LangGraph natively accommodates cycles, which are essential for iterative agent reasoning.

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

The graph relies on three architectural elements:

1. State: A centralized typed data structure that stores the message history, run metadata, and explicit step counters.
2. Nodes: Plain Python functions that receive the current state, run isolated computations (calling the model or executing a tool), and return updated state attributes.
3. Conditional Edges: Decision functions that inspect the state (such as the presence of tool calls or exceeding an iteration ceiling) to decide whether to route to the tool runner or terminate at END.

---

Prerequisites and Dependencies

Ensure your development environment contains the necessary graph orchestration and model driver packages:

pip install langgraph langchain-core langchain-openai

To connect the agent to inference providers through an OpenAI-compatible gateway, configure your environment variables. Using OneMux simplifies unified key management and spend controls across multiple models:

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(Note: Always verify that your selected target model supports structured tool-calling parameters via OpenAI-compatible endpoints).

---

Step 1: Defining Bounded Agent State

In standard conversational workflows, MessagesState handles appending chat history. For coding tasks, we explicitly define AgentState using Python's TypedDict to track message sequences alongside an explicit counter (step_count) and a maximum threshold (max_steps).

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

The add_messages reducer ensures incoming message lists are appended cleanly to conversation history rather than overwriting previous observations.

---

Step 2: Implementing File System and Verification Tools

The agent requires tools to inspect source files, apply updates, and run repository tests. We encapsulate these operations using LangChain's @tool decorator.

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

Step 3: Model Setup via Unified API

We initialize the model using ChatOpenAI, pointing to the unified provider proxy endpoint. Check the OneMux Docs and Quickstart Guide to ensure parameter compatibility for function calling.

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

Step 4: Constructing Graph Nodes and Loop Guards

Nodes execute logic and increment bounds. If an agent loops repeatedly without fixing a broken test, the edge logic detects that step_count >= max_steps and routes directly to graph completion.

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

Step 5: Graph Compilation and Assembly

Assemble the components using StateGraph and compile the workflow:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

Step 6: Executing Code Changes with Test Verification

To run the workflow, define a coding task alongside strict execution bounds:

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

Limitations and Production Hardening

When deploying code-modifying agents in production environments, consider the following safeguards:

• Sandboxing Execution: Running arbitrary shell commands like pytest locally risks environment corruption or unsafe script execution. Isolate test execution within ephemeral Docker containers or secure sandboxes.
• State Token Limits: Extended tool outputs—such as multi-page stack traces—can quickly exhaust context windows. Truncate test output messages before writing them into the state history.
• Managing Model Costs: Iterative ReAct loops multiply token consumption. Track usage via the OneMux Pricing dashboard, and evaluate lower-cost reasoning models on OneMux Models to optimize cost-per-task.

---

Sources

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Updated Sept 2, 2026).