Introduction
Building AI agents capable of handling complex, multi-step tasks requires more than simple chat interfaces. Enter LangGraph, a library from the LangChain ecosystem that enables stateful, cyclical workflows with persistent memory. When paired with Claude from Anthropic, you get an agent architecture leveraging Claude's superior reasoning, tool-calling, and long-context capabilities.
This guide walks you through creating stateful agents using the Claude API and LangGraph. We'll cover setup, basic agents, advanced multi-step examples, and best practices. By the end, you'll have Python code to deploy persistent agents for real-world problems like research, automation, or decision-making.
Why Claude + LangGraph?
- Claude's Strengths: Excellent at multi-step reasoning, XML-structured tool calls, and handling 200K+ token contexts (Claude 3.5 Sonnet).
- LangGraph's Power: Models agent behavior as graphs with nodes (actions/tools), edges (control flow), and checkpointers for state persistence across sessions.
- Use Cases: Research agents, workflow automation, customer support with memory, or any task needing iteration and recall.
Compared to stateless chains, stateful LangGraph agents maintain history, enabling human-in-the-loop interruptions and retries.
Prerequisites
- Python 3.10+
- Anthropic API key (sign up at console.anthropic.com)
- Basic familiarity with LangChain concepts
Step 1: Installation
Install the required packages:
pip install langgraph langchain-anthropic langchain-core python-dotenv
Create a .env file:
ANTHROPIC_API_KEY=your_api_key_here
Step 2: LangGraph Fundamentals
LangGraph uses a StateGraph to define:
- State: A typed dict holding agent memory (e.g., messages, task status).
- Nodes: Functions that update state (e.g., call Claude, execute tools).
- Edges: Conditional routing (e.g.,
should_continue). - Checkpointer: Persists state (e.g.,
MemorySaverfor in-memory, or Postgres for production).
Here's a minimal state schema:
import typing
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
class AgentState(TypedDict):
messages: Annotated[list, "add"]
next: str # For routing
Step 3: Basic Stateful Agent with Claude
Let's build a simple agent that uses Claude to answer questions and call a tool (e.g., calculator).
First, define tools. Claude supports parallel tool calls via the Anthropic API.
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
import operator
@tool
def calculator(expression: str) -> str:
"""Evaluate a math expression."""
try:
return str(eval(expression))
except:
return "Invalid expression"
tools = [calculator]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools)
Now, agent node:
def agent(state: AgentState):
result = llm.invoke(state["messages"])
return {"messages": [result]}
def should_continue(state: AgentState):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
def tool_node(state: AgentState):
outputs = []
for tool_call in state["messages"][-1].tool_calls:
tool = next(t for t in tools if t.name == tool_call["name"])
result = tool.invoke(tool_call["args"])
outputs.append(tool.ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
return {"messages": outputs}
Compile the graph:
workflow = StateGraph(state_schema=AgentState)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
Run with persistent thread:
from dotenv import load_dotenv
load_dotenv()
config = {"configurable": {"thread_id": "abc123"}}
input_messages = [("user", "What is (3 + 5) * 2?")]
for chunk in app.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
This agent remembers across invocations using the same thread_id.
Step 4: Advanced Multi-Step Agent - Research Workflow
For complex tasks, add planning, research (simulated web search), analysis, and reporting. State tracks progress.
Extended state:
class ResearchState(TypedDict):
messages: Annotated[list, operator.add]
plan: str
research_data: list
report: str
next: str
Nodes:
@tool
def web_search(query: str) -> str:
"""Simulate web search (replace with SerpAPI or Tavily)."""
return f"Mock results for '{query}': Key facts from top sources."
tools = [web_search]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620").bind_tools(tools)
# Planning node
def plan(state: ResearchState):
prompt = """Create a step-by-step plan for researching '{topic}'.
Output only the plan as XML: <plan>steps</plan>""".format(topic=state["messages"][-1].content)
result = llm.invoke([("user", prompt)])
return {"plan": result.content, "messages": [result]}
# Research node
def research(state: ResearchState):
plan_steps = state["plan"].split("\
") # Parse steps
queries = [step.split("Search:")[1] for step in plan_steps if "Search:" in step]
data = []
for q in queries[:3]: # Limit
result = web_search.invoke({"query": q.strip()})
data.append(result)
return {"research_data": data, "messages": [("system", f"Gathered data: {data}")]}
# Analyze & Report
def analyze(state: ResearchState):
prompt = f"""Analyze this research data for topic.
Data: {state['research_data']}
Plan: {state['plan']}
Generate a final report."""
result = llm.invoke([("user", prompt)])
return {"report": result.content, "messages": [result], "next": END}
Edges and graph:
def route_research(state: ResearchState):
if not state.get("plan"):
return "plan"
elif not state.get("research_data"):
return "research"
return "analyze"
workflow = StateGraph(state_schema=ResearchState)
workflow.add_node("plan", plan)
workflow.add_node("research", research)
workflow.add_node("analyze", analyze)
workflow.set_entry_point("plan")
workflow.add_conditional_edges("plan", route_research)
workflow.add_conditional_edges("research", route_research)
workflow.add_edge("analyze", END)
app = workflow.compile(checkpointer=MemorySaver())
Stream a research task:
config = {"configurable": {"thread_id": "research1"}}
app.stream({"messages": [("user", "Research the latest on Claude 3.5 Sonnet benchmarks.")]}, config)
# Resume later
print(app.get_state(config).values["report"])
This persists plan, data, and report across sessions.
Step 5: Human-in-the-Loop and Interruptions
Add breakpoints:
from langgraph.checkpoint.sqlite import SqliteSaver
# Use SqliteSaver for production
checkpointer = SqliteSaver.from_conn_string(":memory:")
# In edges, add HumanNode for review
workflow.add_node("human", lambda state: state) # Placeholder
Update config with interrupt_before=["human"] for pauses.
Best Practices for Claude in LangGraph
- Prompt Engineering: Use Claude's XML tags for structure:
<thinking>reason</thinking><action>call</action>. - Tool Calling: Bind tools early; Claude excels at parallel calls (up to 10+).
- State Management: Keep state lean; use summaries for long histories.
- Error Handling: Wrap nodes in try-except, route to "error" node.
- Scaling: Use PostgresSaver for teams; deploy via FastAPI.
- Claude-Specific: Prefer Sonnet for speed/balance; Opus for deepest reasoning. Monitor token usage with
max_tokens. - Testing: Use
app.get_state(config)to inspect persistence.
| Model | Best For | Context |
|---|---|---|
| Haiku | Fast tools | 200K |
| Sonnet | Agents | 200K |
| Opus | Complex plans | 200K |
Production Tips
- Integrate real tools: Tavily for search, DuckDuckGo, or custom APIs.
- Deploy: LangGraph Cloud or self-host with Streamlit/FastAPI.
- Monitor: Log token costs via Anthropic dashboard.
Conclusion
Claude + LangGraph unlocks robust stateful agents for production workflows. Start with the basic calculator example, scale to research pipelines, and iterate with persistence. Fork the code on GitHub, experiment with your tools, and share your agents in the Claude community.
For more: Check Anthropic's API docs and LangGraph guides. Happy building!
(Word count: ~1450)
Stay ahead of the AI curve
The most important updates, news, and content — delivered in one weekly newsletter.