Build a Multi-Step Research Agent with LangGraph, Claude, and Tool Calling

What You'll Build

A stateful research agent that accepts a question, autonomously searches the web and scrapes pages via Claude tool calls, and produces a structured markdown report — wired together in a LangGraph loop that runs until Claude is satisfied.

Prerequisites

Export both keys before running:

export ANTHROPIC_API_KEY="sk-ant-..."
export TAVILY_API_KEY="tvly-..."

1. Install Dependencies

pip install "langgraph>=0.2.0" "anthropic>=0.34.0" tavily-python httpx beautifulsoup4

2. Define Agent State

Create agent.py. LangGraph is a state machine — every node receives the full state and returns a partial update that gets merged automatically.

import os
import anthropic
import httpx
from bs4 import BeautifulSoup
from tavily import TavilyClient
from langgraph.graph import StateGraph, END, START
from typing import TypedDict

claude = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY automatically
tavily = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

class AgentState(TypedDict):
    messages: list    # Claude-format message dicts
    stop_reason: str  # "tool_use" | "end_turn"
    report: str       # final compiled report

3. Implement the Tools

def web_search(query: str) -> str:
    results = tavily.search(query=query, max_results=3)
    return "\n\n".join(
        f"[{r['title']}]({r['url']})\n{r['content']}"
        for r in results["results"]
    )

def scrape_page(url: str) -> str:
    try:
        resp = httpx.get(url, timeout=10, follow_redirects=True,
                         headers={"User-Agent": "ResearchBot/1.0"})
        soup = BeautifulSoup(resp.text, "html.parser")
        for tag in soup(["script", "style", "nav", "footer"]):
            tag.decompose()
        return soup.get_text(separator="\n", strip=True)[:4000]
    except Exception as e:
        return f"Error scraping {url}: {e}"

def run_tool(name: str, inputs: dict) -> str:
    if name == "web_search":
        return web_search(inputs["query"])
    if name == "scrape_page":
        return scrape_page(inputs["url"])
    return f"Unknown tool: {name}"

The scraper caps output at 4,000 characters — enough detail without overrunning Claude's context window.

4. Register Tools with Claude

Anthropic's tool calling requires a JSON Schema input_schema per tool.

TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web for recent information on a topic.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    },
    {
        "name": "scrape_page",
        "description": "Fetch and extract readable text content from a URL.",
        "input_schema": {
            "type": "object",
            "properties": {"url": {"type": "string"}},
            "required": ["url"]
        }
    }
]

SYSTEM = (
    "You are a research assistant. Use web_search and scrape_page to gather information. "
    "When you have enough, respond with a structured markdown report and no further tool calls."
)

5. Build the Graph Nodes

def agent(state: AgentState) -> dict:
    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        system=SYSTEM,
        tools=TOOLS,
        messages=state["messages"]
    )
    return {
        "messages": state["messages"] + [
            {"role": "assistant", "content": response.content}
        ],
        "stop_reason": response.stop_reason,
    }

def tool_executor(state: AgentState) -> dict:
    last_msg = state["messages"][-1]
    tool_results = []
    for block in last_msg["content"]:   # SDK ContentBlock objects
        if block.type == "tool_use":
            print(f"[tool] {block.name}({block.input})")
            output = run_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output
            })
    return {
        "messages": state["messages"] + [
            {"role": "user", "content": tool_results}
        ],
        "stop_reason": "",
    }

def extract_report(state: AgentState) -> dict:
    text = "".join(
        block.text
        for block in state["messages"][-1]["content"]
        if block.type == "text"
    )
    return {"report": text}

def router(state: AgentState) -> str:
    return "tool_executor" if state["stop_reason"] == "tool_use" else "extract_report"

Two notes on the SDK behavior:

• response.content is a list of TextBlock/ToolUseBlock objects. Passing it directly to the next messages.create call is the documented pattern — the SDK serializes them correctly.
• All tool results for one round must be batched into a single user message; the list in tool_executor handles this automatically.

6. Wire the Graph

builder = StateGraph(AgentState)
builder.add_node("agent", agent)
builder.add_node("tool_executor", tool_executor)
builder.add_node("extract_report", extract_report)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", router)  # → "tool_executor" or "extract_report"
builder.add_edge("tool_executor", "agent")       # loop back
builder.add_edge("extract_report", END)

research_agent = builder.compile()

The cycle agent → tool_executor → agent repeats until stop_reason is "end_turn", then the graph exits through extract_report.

Advertisement

Serverless Inference by DigitalOcean 55+ models, every modality. One API key, one bill.

7. Run the Agent

Append to agent.py:

if __name__ == "__main__":
    topic = "What are the most significant AI safety developments in 2024?"
    result = research_agent.invoke({
        "messages": [{"role": "user", "content": topic}],
        "stop_reason": "",
        "report": ""
    })
    print(result["report"])

python agent.py

Verify It Works

Before the final report prints, you should see tool log lines:

[tool] web_search({'query': 'AI safety 2024 developments'})
[tool] scrape_page({'url': 'https://...'})
[tool] web_search({'query': 'AI safety regulations 2024'})

Followed by a markdown report in result["report"]. If the [tool] lines never appear, Claude received the messages but the state graph is not looping — double-check add_conditional_edges is present.

Troubleshooting

KeyError: 'results' from Tavily — Your TAVILY_API_KEY is missing or invalid. Free tier also enforces rate limits; add time.sleep(1) between heavy runs.

AttributeError: 'dict' object has no attribute 'type' — You serialized content blocks with .model_dump() or dict() before the executor read them. Keep response.content as-is; never convert it manually.

Agent loops forever — Claude needs a clearer stopping signal. Strengthen the system prompt, or add a recursion cap: research_agent.invoke(initial_state, {"recursion_limit": 12}).

BadRequestError: roles must alternate — Two consecutive messages of the same role were added. This usually means tool_result blocks were added as separate user messages instead of one batched list — the tool_results list in tool_executor prevents this if kept intact.

Next Steps

• Persistence — Pass checkpointer=MemorySaver() to builder.compile() to resume interrupted runs by thread ID.
• Streaming — Replace invoke with research_agent.stream(...) and iterate events for real-time token output.
• Additional tools — Add a read_pdf or run_python tool following the same input_schema pattern; run_tool dispatches by name.
• Observability — Set LANGCHAIN_TRACING_V2=true and LANGCHAIN_API_KEY to get a full visual trace of every node and tool call in LangSmith.