Chapter 3 of 8
Building Custom AI Agent Loops in Python
Build complete custom agent loops using LangChain, CrewAI, and the Anthropic SDK — with real code examples.
Python is the dominant language for building AI agent loops, thanks to its mature LLM ecosystem and the wide availability of agent frameworks. This tutorial walks you through three production-tested approaches: using the raw Anthropic SDK for maximum control, LangChain/LangGraph for graph-based workflows, and CrewAI for multi-agent collaboration. Each approach includes working code you can run today.
The open-source ecosystem has matured rapidly. Frameworks like LangGraph (github.com/langchain-ai/langgraph) provide graph-based orchestration, CrewAI (github.com/crewAIInc/crewAI) enables team-based multi-agent workflows, and tools like Aider (github.com/paul-gauthier/aider, 30K+ GitHub stars) demonstrate how agent loops work in production CLI tools. Production coding agents like Claude Code (github.com/anthropics/claude-code) and OpenHands (github.com/All-Hands-AI/OpenHands) all implement variations of the same core loop pattern described below.
The Core Loop Pattern
Every AI agent loop follows the same fundamental cycle, regardless of which framework you use:
+----------+
| PLAN | Agent decides what to do next
+----+-----+
|
+----v-----+
| ACT | Agent calls a tool or generates text
+----+-----+
|
+----v-----+
| OBSERVE | Tool returns results to the agent
+----+-----+
|
+----v-----+
| REFLECT | Agent evaluates: am I done?
+----+-----+
|
+----v-----+
| Done? |--- Yes ---> Return result
+----+-----+
|
No
|
+-------> Back to PLAN
The loop continues until a stopping condition is met: the agent signals task completion, a maximum iteration limit is reached, or an error threshold is exceeded.
Approach 1: Raw SDK Loop (Maximum Control)
The most educational approach is building a loop from scratch using the Anthropic SDK directly. This teaches you exactly how tool calling works at the protocol level -- the same pattern used internally by Claude Code, Codex CLI (github.com/openai/codex), and Aider.
Install the SDK:
pip install anthropic
Here is a complete, working agent loop with tool calling:
"""minimal_agent_loop.py - A minimal agent loop using the Anthropic SDK."""
import anthropic
import json
client = anthropic.Anthropic()
# Define tools the agent can use
tools = [
{
"name": "read_file",
"description": "Read the contents of a file from the filesystem.",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to read."
}
},
"required": ["path"]
}
},
{
"name": "search_files",
"description": "Search for files matching a pattern in a directory.",
"input_schema": {
"type": "object",
"properties": {
"directory": {
"type": "string",
"description": "The directory to search in."
},
"pattern": {
"type": "string",
"description": "The filename pattern to match (glob)."
}
},
"required": ["directory", "pattern"]
}
},
{
"name": "run_command",
"description": "Run a shell command and return the output.",
"input_schema": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to run."
}
},
"required": ["command"]
}
}
]
# Tool implementation functions
def read_file(path: str) -> str:
"""Read a file and return its contents."""
try:
with open(path, "r") as f:
return f.read()
except FileNotFoundError:
return f"Error: File not found: {path}"
def search_files(directory: str, pattern: str) -> str:
"""Search for files matching a pattern."""
import glob
matches = glob.glob(f"{directory}/{pattern}")
return json.dumps(matches) if matches else "No files found."
def run_command(command: str) -> str:
"""Run a shell command."""
import subprocess
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
return result.stdout or result.stderr
# Map tool names to functions
TOOL_FUNCTIONS = {
"read_file": read_file,
"search_files": search_files,
"run_command": run_command,
}
def run_agent_loop(task: str, max_iterations: int = 10):
"""Run an agent loop with tool calling until the task is complete."""
messages = [{"role": "user", "content": task}]
for i in range(max_iterations):
print(f"\n--- Iteration {i + 1} ---")
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=tools,
messages=messages
)
# Add the assistant's response to the conversation
messages.append({"role": "assistant", "content": response.content})
# Check if the agent wants to call any tools
tool_uses = [block for block in response.content if block.type == "tool_use"]
if not tool_uses:
# No tool calls -- the agent is done
print("Agent completed the task.")
return response.content
# Execute each tool call and feed results back
for tool_use in tool_uses:
tool_name = tool_use.name
tool_input = tool_use.input
tool_id = tool_use.id
print(f" Tool call: {tool_name}({tool_input})")
# Execute the tool
if tool_name in TOOL_FUNCTIONS:
result = TOOL_FUNCTIONS[tool_name](**tool_input)
else:
result = f"Error: Unknown tool '{tool_name}'"
# Feed the tool result back to the agent
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_id,
"content": result
}]
})
print(f"Reached max iterations ({max_iterations}) without completion.")
return response.content
if __name__ == "__main__":
result = run_agent_loop(
task="Read the README.md file and list all API endpoints mentioned in it.",
max_iterations=5
)
How this works: The agent receives a task, decides whether to call a tool, the tool result is fed back as a tool_result message, and the loop continues. When the agent stops calling tools, the loop ends. This is the exact same pattern that Aider implements in its git-first CLI workflow (aider --model claude-3.5-sonnet), and that Claude Code uses under the hood for tool-driven code editing.
Approach 2: LangChain with LangGraph
Install the dependencies:
pip install langchain langchain-anthropic langgraph
"""langgraph_agent.py - An agent loop using LangGraph with tool calling."""
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
# Define tools using the @tool decorator
@tool
def read_file(path: str) -> str:
"""Read the contents of a file from the filesystem."""
with open(path, "r") as f:
return f.read()
@tool
def search_code(query: str, directory: str = ".") -> str:
"""Search for a string pattern in Python files within a directory."""
import subprocess
result = subprocess.run(
["grep", "-r", "-n", query, directory, "--include=*.py"],
capture_output=True, text=True
)
return result.stdout or "No matches found."
@tool
def list_issues() -> str:
"""List open GitHub issues for the current repository."""
import subprocess
result = subprocess.run(
["gh", "issue", "list", "--state", "open", "--limit", "10"],
capture_output=True, text=True
)
return result.stdout or "No issues found or gh CLI not configured."
# Create the model
model = ChatAnthropic(
model="claude-sonnet-4-20250514",
max_tokens=4096
)
# Build the agent with create_react_agent
# LangGraph handles the loop: model -> tool call -> tool result -> model -> ...
agent = create_react_agent(
model=model,
tools=[read_file, search_code, list_issues]
)
# Run the agent
result = agent.invoke({
"messages": [
{"role": "user", "content": "Find all TODO comments in the src/ directory and summarize them."}
]
})
# Print the final response
for msg in result["messages"]:
if hasattr(msg, "content") and msg.content:
print(f"[{msg.type}]: {msg.content[:200]}")
Key advantages of LangGraph:
- Built-in loop management --
create_react_agentimplements the Plan-Act-Observe-Reflect cycle automatically - State management -- LangGraph maintains conversation state across iterations via a strongly-typed state graph
- Checkpointing -- Built-in state persistence for long-running loops (SQLite, Postgres)
- Max iterations -- Set with
recursion_limitparameter to prevent infinite loops
Approach 3: CrewAI (Multi-Agent Collaboration)
Install the dependency:
pip install crewai
Here is a practical code review loop using CrewAI:
"""crewai_code_review.py - A code review loop using CrewAI."""
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
@tool
def read_source_file(filepath: str) -> str:
"""Read a source code file for review."""
with open(filepath, "r") as f:
return f.read()
@tool
def write_source_file(filepath: str, content: str) -> str:
"""Write updated source code to a file."""
with open(filepath, "w") as f:
f.write(content)
return f"Wrote {len(content)} characters to {filepath}"
@tool
def run_tests(test_command: str) -> str:
"""Run a test command and return the results."""
import subprocess
result = subprocess.run(test_command, shell=True, capture_output=True, text=True, timeout=60)
return f"Exit code: {result.returncode}\nstdout: {result.stdout}\nstderr: {result.stderr}"
# Define agents with distinct roles
developer = Agent(
role="Senior Python Developer",
goal="Write clean, well-tested code that meets the requirements",
backstory="You are an experienced developer who writes idiomatic Python "
"with type hints, docstrings, and comprehensive error handling. "
"You always write tests for your code.",
tools=[read_source_file, write_source_file, run_tests],
verbose=True
)
reviewer = Agent(
role="Code Reviewer",
goal="Review code for bugs, style issues, and test coverage",
backstory="You are a meticulous code reviewer who checks for "
"security vulnerabilities, performance issues, "
"PEP 8 compliance, and adequate test coverage. "
"You provide specific, actionable feedback.",
tools=[read_source_file, run_tests],
verbose=True
)
# Define tasks
implement_task = Task(
description="Implement a function `merge_sorted_lists(a: list, b: list) -> list` "
"that merges two sorted lists into one sorted list. Write the function "
"in src/merge.py and add tests in tests/test_merge.py. Run the tests "
"to verify the implementation.",
agent=developer,
expected_output="A working merge_sorted_lists function with passing tests."
)
review_task = Task(
description="Review the code in src/merge.py and tests/test_merge.py. "
"Check for: correctness, edge cases (empty lists, duplicates), "
"performance (should be O(n+m)), PEP 8 compliance, and test coverage. "
"If issues are found, list them specifically so the developer can fix them.",
agent=reviewer,
expected_output="A list of issues found, or a pass if the code is acceptable."
)
fix_task = Task(
description="Address all issues raised in the code review. Fix the code, "
"update the tests if needed, and re-run the test suite to confirm "
"all tests pass.",
agent=developer,
expected_output="Updated code with all review issues resolved and tests passing."
)
# Build and run the crew (this is the loop)
# CrewAI iterates through tasks, and agents collaborate automatically
code_review_crew = Crew(
agents=[developer, reviewer],
tasks=[implement_task, review_task, fix_task],
process=Process.sequential, # Tasks run in order
verbose=True
)
result = code_review_crew.run()
print("\n--- Final Result ---")
print(result)
How CrewAI's loop works: Each task is assigned to an agent. When the reviewer finds issues, the developer agent gets another task to fix them. This back-and-forth is the "loop" -- the crew iterates until the quality gate passes. You can extend this with more iterations by adding conditional logic or using Process.hierarchical mode.
For comparison, Microsoft AutoGen (github.com/microsoft/autogen) offers a similar multi-agent approach but with a conversation-centric model, while MetaGPT (github.com/geekan/MetaGPT, 45K+ GitHub stars) implements a full software company simulation with role-defined agents. A framework comparison provides detailed architecture and performance benchmarks across LangGraph, CrewAI, and AutoGen for enterprise deployment scenarios.
Adding Tools (Function Calling)
Regardless of which approach you choose, tools follow the same pattern. The LLM generates a structured tool call (function name + JSON arguments), your code executes the function, and the result is fed back to the LLM. This is the same mechanism used by Cline (github.com/cline/cline), the VS Code plugin that supports MCP (Model Context Protocol) for tool integration.
A well-defined tool has:
- A clear name that describes what it does (
read_file, nottool1) - A description that the LLM uses to decide when to call it
- A JSON schema defining the input parameters
- Error handling that returns useful error messages back to the agent
Common tool categories for agent loops:
| Category | Example Tools | Use Case |
|---|---|---|
| File operations | read_file, write_file, search_files | Code editing, document processing |
| Shell commands | run_command, run_tests | Build, test, deploy workflows |
| Web search | search_web, fetch_url | Research, fact-checking |
| API calls | create_issue, query_database | Integration with external systems |
| Code analysis | lint_code, type_check | Quality enforcement |
State Management Across Iterations
There are three practical approaches:
In-memory (simplest): Keep a list of messages in Python. Works for single-session loops. Lost if the process crashes.
# In-memory message history
messages = []
messages.append({"role": "user", "content": initial_task})
messages.append({"role": "assistant", "content": response})
# This grows each iteration -- manage with pruning (see Token Overload guide)
File-based (practical): Serialize state to JSON after each iteration. Survives restarts.
import json
from pathlib import Path
def save_state(state: dict, path: str = "loop_state.json"):
Path(path).write_text(json.dumps(state, indent=2))
def load_state(path: str = "loop_state.json") -> dict | None:
p = Path(path)
if p.exists():
return json.loads(p.read_text())
return None
LangGraph checkpointing (production): LangGraph has built-in checkpointing that persists the full graph state to a database (SQLite, Postgres). This is the recommended approach for production loops.
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("agent_state.db")
agent = create_react_agent(model, tools, checkpointer=checkpointer)
# Resume from the last checkpoint
config = {"configurable": {"thread_id": "my-loop-001"}}
result = agent.invoke({"messages": [task]}, config)
For more on state management, see State Persistence.
Preventing Infinite Loops
Always implement these safeguards:
# 1. Max iteration limit (most important)
MAX_ITERATIONS = 10
# 2. Loop detection: track which tools were called
recent_tool_calls = []
if tool_name in recent_tool_calls[-3:]:
print(f"WARNING: Agent called {tool_name} repeatedly. Stopping.")
break
recent_tool_calls.append(tool_name)
# 3. Token budget cap
if total_tokens_spent > MAX_TOKEN_BUDGET:
print("Token budget exceeded. Stopping loop.")
break
# 4. Cost budget cap
if total_cost_usd > MAX_COST_USD:
print("Cost budget exceeded. Stopping loop.")
break
# 5. Time budget cap
import time
if time.time() - start_time > MAX_WALL_TIME_SECONDS:
print("Wall time exceeded. Stopping loop.")
break
The same loop detection pattern is used by Aider in its git-first approach -- each iteration creates a git commit, making it trivial to detect and roll back infinite loops. Claude Code implements similar safeguards via its recursion_limit and per-tool timeout configurations.
Choosing the Right Approach
| Factor | Raw SDK | LangGraph | CrewAI |
|---|---|---|---|
| Complexity | Low | Medium | Low |
| Control | Maximum | High | Medium |
| Multi-agent | Manual setup | Manual setup | Built-in |
| State persistence | DIY | Built-in checkpointing | Basic |
| Best for | Learning, simple loops | Complex workflows | Team-style collaboration |
| Framework overhead | None | Medium | Low |
Next Steps
- For handling the timeouts that occur in long-running loops, see Solving Loop Timeout Issues
- For managing token costs in iterative loops, see Preventing Token Overload
- For multi-agent architectures, see Multi-Agent Loop
- For setting up your local environment, see Local Loop Workflow Setup