Introduction

Building AI agents that maintain state across sessions, recover from failures, and execute intricate multi-step reasoning is crucial for real-world applications. Claude, with its superior reasoning and long-context capabilities, paired with LangGraph's graph-based orchestration, delivers exactly that: persistent, resilient agents.

In this guide, we'll walk through creating such agents step-by-step. Expect practical Python code, Claude-specific optimizations, and examples like a research workflow that checkpoints progress and retries on errors.

Why Claude + LangGraph?

LangGraph, from the LangChain ecosystem, models agent logic as graphs with nodes (actions/tools) and edges (transitions). It shines for stateful, cyclical workflows.

Claude excels here due to:

• Superior reasoning: Handles complex chains better than shorter-context models.
• Tool use: Native support via Messages API for structured outputs.
• Long context: Up to 200K tokens in Opus, ideal for memory-heavy agents.

Benefits include:

• Persistence: Checkpoints save state to resume later.
• Error recovery: Built-in retries and human-in-loop.
• Scalability: Deploy as APIs for production workflows.

Prerequisites

• Python 3.10+
• Anthropic API key (from console.anthropic.com)
• Familiarity with async Python and LangChain basics

Step 1: Install Dependencies

pip install langgraph langchain-anthropic python-dotenv

Create a .env file:

ANTHROPIC_API_KEY=your_key_here

Step 2: Set Up Claude Client

import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic

load_dotenv()
claude = ChatAnthropic(
    model="claude-3-5-sonnet-20240620",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    temperature=0.1
)

Sonnet balances speed and reasoning; swap to Opus for deeper tasks.

Step 3: Define Agent State

State tracks messages, memory, and custom fields:

from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[List[dict], add_messages]
    research_summary: str
    next_step: str

This enables persistent memory across runs.

Step 4: Build Core Nodes

Nodes are Claude-powered functions. Example: researcher node.

async def researcher(state: AgentState) -> AgentState:
    msg = state["messages"][-1]["content"]
    prompt = f"""
    Research '{msg}'. Provide a summary and next action.
    Output JSON: {{"summary": "...", "next_step": "search|analyze|finish"}}
    """
    response = await claude.ainvoke(prompt)
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "research_summary": response.content,  # Parse JSON in prod
        "next_step": "search"  # Simplified
    }

Add tools node:

def tools_node(state: AgentState):
    # Simulate tool calls (e.g., web search)
    return {"messages": [{"role": "tool", "content": "Search results..."}]}

Step 5: Construct the Graph

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

workflow = StateGraph(AgentState)

workflow.add_node("researcher", researcher)
workflow.add_node("tools", tools_node)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "tools")
workflow.add_conditional_edges(
    "tools",
    lambda s: s["next_step"],
    {"search": "researcher", "finish": END}
)
workflow.add_edge("researcher", END)  # Simplified

# Compile without persistence yet
graph = workflow.compile()

Step 6: Add Persistent Memory with Checkpointers

Persistence via MemorySaver (in-memory) or Postgres for prod.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# Run with thread_id for session persistence
config = {"configurable": {"thread_id": "agent_1"}}

input_message = {"messages": [{"role": "user", "content": "Research quantum computing"}]}

for chunk in graph.stream(input_message, config, stream_mode="values"):
    print(chunk)

Resume later:

# Same config resumes from checkpoint
resumed_input = {"messages": [{"role": "user", "content": "Continue research"}]}
graph.stream(resumed_input, config)

Claude's context window preserves full history effortlessly.

Step 7: Implement Error Recovery

Wrap nodes in retries:

import asyncio

async def robust_researcher(state: AgentState) -> AgentState:
    for attempt in range(3):
        try:
            return await researcher(state)
        except Exception as e:
            if attempt == 2:
                raise
            await asyncio.sleep(2 ** attempt)
    return state

# Or use LangGraph's built-in retry_policy
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["researcher"],
    retry_policy={"researcher": {"max_attempts": 3}}
)

Human-in-loop: Interrupt on error, inspect graph.get_state(config), then resume.

Real-World Example: Multi-Step Research Agent

Full workflow for topic research: search → summarize → analyze → report.

class ResearchState(TypedDict):
    messages: Annotated[List[dict], add_messages]
    summary: str
    analysis: str
    status: str  # 'searching|summarizing|analyzing|done'

# Nodes
async def search_node(state):
    # Fake Tavily search or real integration
    return {"summary": "Quantum bits enable superposition...", "status": "summarizing"}

async def summarize_node(state):
    prompt = f"Summarize: {state['summary']}"
    resp = await claude.ainvoke(prompt)
    return {"messages": [resp], "status": "analyzing"}

async def analyze_node(state):
    prompt = f"Analyze implications: {state['summary']}"
    resp = await claude.ainvoke(prompt)
    return {"analysis": resp.content, "status": "done"}

# Graph
research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("summarize", summarize_node)
research_graph.add_node("analyze", analyze_node)

research_graph.set_entry_point("search")
research_graph.add_edge("search", "summarize")
research_graph.add_edge("summarize", "analyze")
research_graph.add_edge("analyze", END)

checkpointer = MemorySaver()
research_agent = research_graph.compile(checkpointer=checkpointer)

# Usage
config = {"configurable": {"thread_id": "research_1"}}
result = await research_agent.ainvoke(
    {"messages": [{"role": "user", "content": "Start quantum research"}]},
    config
)
print(result["analysis"])

This checkpoints after each step—pause mid-research, resume anytime.

Step 8: Deploy for Production Workflows

• API Server: Use FastAPI + LangGraph Server.

from langgraph.deploy.fastapi import create_app
app = create_app(graph)

• Integrations: Hook to n8n/Zapier via webhooks.
• Scaling: PostgresSaver for SQLite/Postgres checkpointers.

Best Practices for Claude + LangGraph

1. Prompt Engineering: Use XML tags for Claude: <thinking>Reason step-by-step</thinking>.
2. State Pruning: Compress old messages with Claude summarization node.
3. Model Selection: Haiku for fast tools, Sonnet/Opus for reasoning.
4. Monitoring: Log checkpoints, track token usage.
5. Security: Validate tool inputs, rate-limit API calls.
6. Testing: Unit test nodes, simulate failures.
7. Cost Optimization: Cache common subgraphs.

Common Pitfalls and Fixes

• State Bloat: Implement a "reflect" node to summarize history.
• Infinite Loops: Add max iterations in edges.
• Tool Errors: Claude's structured outputs prevent parsing fails.

Conclusion

Claude + LangGraph empowers agents that think like teams: persistent, recoverable, and smart. Start with the research example, adapt to HR onboarding, sales pipelines, or code reviews.

Experiment in Colab, deploy to prod. Share your graphs on Claude Directory forums!