Building an Autonomous Research Agent with Claude

A Claude research agent takes a question, runs multiple web searches, reads source documents, deduplicates findings, and produces a structured report — without manual intervention at each step. A well-architected agent can process 10–15 sources and deliver a 500-word cited summary in under 60 seconds, at a cost of roughly $0.05–$0.15 per research session using Claude Sonnet 3.7.

Architecture: Four-Component Pattern

The most reliable research agent architecture separates concerns into four sequential components:

Question
    │
    ▼
[Planner]  → generates a search plan (3–5 queries)
    │
    ▼
[Searcher] → executes searches, fetches pages (parallel)
    │
    ▼
[Synthesizer] → reads sources, extracts key facts, tracks citations
    │
    ▼
[Formatter] → assembles structured report with inline citations
    │
    ▼
Report

Each component is a Claude call with a specific system prompt and tool set. This separation makes the agent easier to debug (swap one component at a time), cheaper to run (use Haiku for search planning, Sonnet for synthesis), and more reliable than a single monolithic agent loop.

Tool Definitions

Define four tools that cover the agent's capabilities. These tool schemas should be cached (see the prompt caching guide) since they don't change between sessions.

import anthropic
import httpx
from bs4 import BeautifulSoup
from typing import Any

client = anthropic.Anthropic()

# Tool schemas
RESEARCH_TOOLS = [
    {
        "name": "web_search",
        "description": """Search the web for current information on a topic. 
        Returns a list of results with titles, URLs, and snippets.
        Use specific, targeted queries. Prefer queries with named entities, dates, or version numbers.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search query string (be specific)"
                },
                "num_results": {
                    "type": "integer",
                    "description": "Number of results to return (1-10)",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "read_page",
        "description": """Fetch and read the full text content of a web page.
        Use this after web_search to get the full content of a promising result.
        Returns cleaned text with the source URL for citation.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "Full URL of the page to read"
                },
                "max_chars": {
                    "type": "integer",
                    "description": "Maximum characters to return (default 4000)",
                    "default": 4000
                }
            },
            "required": ["url"]
        }
    },
    {
        "name": "extract_facts",
        "description": """Extract and store a specific fact or claim from a source.
        Call this for each distinct claim you want to include in the final report.
        Facts are deduplicated automatically — duplicate claims from different sources 
        will be merged with all source URLs retained.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "fact": {
                    "type": "string",
                    "description": "The factual claim to store"
                },
                "source_url": {
                    "type": "string",
                    "description": "URL where this fact was found"
                },
                "source_title": {
                    "type": "string",
                    "description": "Title of the source page"
                },
                "confidence": {
                    "type": "string",
                    "enum": ["high", "medium", "low"],
                    "description": "Confidence in this fact"
                }
            },
            "required": ["fact", "source_url", "confidence"]
        }
    },
    {
        "name": "write_report",
        "description": """Compile all extracted facts into a structured research report.
        Call this once after you have gathered sufficient facts (minimum 5).
        The report will include an executive summary, key findings, and a sources list.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {
                    "type": "string",
                    "description": "Report title"
                },
                "executive_summary": {
                    "type": "string",
                    "description": "2-3 sentence summary of findings"
                },
                "format": {
                    "type": "string",
                    "enum": ["markdown", "json", "plain"],
                    "default": "markdown"
                }
            },
            "required": ["title", "executive_summary"]
        }
    }
]

Tool Implementations

import json
import re
from urllib.parse import quote_plus

# In-memory fact store with deduplication
fact_store: list[dict] = []

def execute_tool(tool_name: str, tool_input: dict) -> str:
    if tool_name == "web_search":
        return web_search(tool_input["query"], tool_input.get("num_results", 5))
    elif tool_name == "read_page":
        return read_page(tool_input["url"], tool_input.get("max_chars", 4000))
    elif tool_name == "extract_facts":
        return extract_fact(
            tool_input["fact"],
            tool_input["source_url"],
            tool_input.get("source_title", ""),
            tool_input["confidence"]
        )
    elif tool_name == "write_report":
        return compile_report(
            tool_input["title"],
            tool_input["executive_summary"],
            tool_input.get("format", "markdown")
        )
    return f"Unknown tool: {tool_name}"


def web_search(query: str, num_results: int = 5) -> str:
    """
    In production: use Brave Search API, SerpAPI, or Tavily.
    Here we stub with DuckDuckGo Instant Answer API for illustration.
    """
    try:
        url = f"https://api.search.brave.com/res/v1/web/search?q={quote_plus(query)}&count={num_results}"
        # Replace with your actual search API key/endpoint
        headers = {"Accept": "application/json", "X-Subscription-Token": "YOUR_BRAVE_API_KEY"}
        resp = httpx.get(url, headers=headers, timeout=10)
        data = resp.json()
        results = []
        for r in data.get("web", {}).get("results", [])[:num_results]:
            results.append({
                "title": r.get("title", ""),
                "url": r.get("url", ""),
                "snippet": r.get("description", "")
            })
        return json.dumps(results, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def read_page(url: str, max_chars: int = 4000) -> str:
    """Fetch a URL and return cleaned text content."""
    try:
        resp = httpx.get(url, timeout=15, follow_redirects=True,
                        headers={"User-Agent": "ResearchBot/1.0"})
        soup = BeautifulSoup(resp.text, "html.parser")
        # Remove nav, footer, ads
        for tag in soup(["script", "style", "nav", "footer", "aside", "iframe"]):
            tag.decompose()
        text = soup.get_text(separator="\n", strip=True)
        # Collapse whitespace
        text = re.sub(r"\n{3,}", "\n\n", text)
        return json.dumps({"url": url, "content": text[:max_chars]})
    except Exception as e:
        return json.dumps({"error": str(e), "url": url})


def extract_fact(fact: str, source_url: str, source_title: str, confidence: str) -> str:
    """Store fact with deduplication — merge sources for near-duplicate facts."""
    global fact_store
    
    # Simple deduplication: check if we already have a very similar fact
    for existing in fact_store:
        # If >80% word overlap, treat as duplicate and add source
        existing_words = set(existing["fact"].lower().split())
        new_words = set(fact.lower().split())
        overlap = len(existing_words & new_words) / max(len(existing_words | new_words), 1)
        if overlap > 0.8:
            if source_url not in existing["sources"]:
                existing["sources"].append({"url": source_url, "title": source_title})
            return json.dumps({"status": "merged", "fact_id": existing["id"]})
    
    fact_id = len(fact_store)
    fact_store.append({
        "id": fact_id,
        "fact": fact,
        "confidence": confidence,
        "sources": [{"url": source_url, "title": source_title}]
    })
    return json.dumps({"status": "stored", "fact_id": fact_id})


def compile_report(title: str, executive_summary: str, fmt: str = "markdown") -> str:
    """Compile stored facts into a structured report."""
    high = [f for f in fact_store if f["confidence"] == "high"]
    medium = [f for f in fact_store if f["confidence"] == "medium"]
    low = [f for f in fact_store if f["confidence"] == "low"]
    
    # Collect unique sources
    all_sources = {}
    for fact in fact_store:
        for src in fact["sources"]:
            all_sources[src["url"]] = src.get("title", src["url"])
    
    if fmt == "markdown":
        lines = [f"# {title}\n", f"{executive_summary}\n"]
        if high:
            lines.append("## Key Findings\n")
            for f in high:
                src_refs = ", ".join(f"[{s.get('title', s['url'])[:40]}]({s['url']})" 
                                      for s in f["sources"])
                lines.append(f"- {f['fact']} ({src_refs})")
        if medium:
            lines.append("\n## Additional Findings\n")
            for f in medium:
                lines.append(f"- {f['fact']}")
        lines.append("\n## Sources\n")
        for url, title_str in all_sources.items():
            lines.append(f"- [{title_str}]({url})")
        return "\n".join(lines)
    
    return json.dumps({
        "title": title,
        "summary": executive_summary,
        "high_confidence_facts": high,
        "medium_confidence_facts": medium,
        "sources": list(all_sources.keys())
    }, indent=2)

The Agent Loop

RESEARCH_SYSTEM_PROMPT = """You are an autonomous research agent. Your job is to:

1. PLAN: Identify 3-5 distinct search queries that will comprehensively answer the research question
2. SEARCH: Execute all planned searches using web_search
3. READ: For the most relevant results, use read_page to get full content  
4. EXTRACT: Use extract_facts for every distinct claim you find (aim for 8-15 facts)
5. REPORT: Call write_report once with a concise executive summary

Guidelines:
- Be systematic: complete all searches before writing the report
- Prefer primary sources (official docs, announcements) over secondary sources
- Mark confidence as "high" only for facts from official/primary sources
- Always include the source URL when extracting facts
- Stop after 20 tool calls to control costs

Do not ask for clarification. Work autonomously until the report is complete."""


def run_research_agent(question: str, max_tool_calls: int = 20) -> str:
    """Run the research agent and return the final report."""
    global fact_store
    fact_store = []  # Reset for new session
    
    messages = [{"role": "user", "content": question}]
    tool_calls_used = 0
    final_report = None
    
    while tool_calls_used < max_tool_calls:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            system=[
                {
                    "type": "text",
                    "text": RESEARCH_SYSTEM_PROMPT,
                    "cache_control": {"type": "ephemeral"}  # Cache system prompt
                }
            ],
            tools=[
                {**tool, **({"cache_control": {"type": "ephemeral"}} 
                             if i == len(RESEARCH_TOOLS) - 1 else {})}
                for i, tool in enumerate(RESEARCH_TOOLS)
            ],
            messages=messages
        )
        
        # Add assistant response to history
        messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason == "end_turn":
            break
        
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_calls_used += 1
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
                    
                    # Capture the final report when write_report is called
                    if block.name == "write_report":
                        # Re-run compile with same args to get the output
                        final_report = result
            
            messages.append({"role": "user", "content": tool_results})
        
        # Cost guard: if approaching limit, signal the agent to wrap up
        if tool_calls_used >= max_tool_calls - 2:
            messages.append({
                "role": "user",
                "content": "You have 2 tool calls remaining. Call write_report now to complete the research."
            })
    
    return final_report or compile_report(
        "Research Results", 
        f"Research on: {question}", 
        "markdown"
    )

Example: Researching Claude API Pricing Changes 2025–2026

question = """Research: What changes did Anthropic make to Claude API pricing 
between 2025 and 2026? Include specific price points, model names, and dates 
where available."""

report = run_research_agent(question)
print(report)

Expected output (abbreviated):

# Claude API Pricing Changes 2025–2026

Anthropic reduced Claude API prices significantly in 2025–2026, with the
introduction of Claude 3.5 Haiku as a low-cost tier and continued price
reductions across Sonnet and Opus tiers. Prompt caching availability expanded
to all models, providing up to 90% savings on repeated context.

## Key Findings

- Claude 3.5 Haiku launched at $0.80/$4 per million input/output tokens, 
  the lowest price in the Claude lineup ([Anthropic Blog](https://anthropic.com/...))
- Claude Sonnet 3.7 input price: $3.00/M tokens; cache read: $0.30/M tokens
  ([Anthropic Pricing](https://anthropic.com/api))
- Batch API discount: 50% off all model prices for asynchronous workloads
  ([Batch API Docs](https://docs.anthropic.com/...))

## Sources

- [Anthropic API Pricing](https://anthropic.com/api)
- [Model Comparison — Anthropic](https://docs.anthropic.com/en/docs/models)

A 10-source research session like this typically costs $0.08–$0.15 with Sonnet 3.7 and prompt caching enabled.

Cost Controls for Long Research Sessions

Research agents can spiral in cost if not bounded. Three practical controls:

1. Hard tool call cap. Set max_tool_calls=20 and pass a "wrap up" signal when 2 calls remain. This bounds your worst-case cost to roughly 20 × (avg tokens per call) × price.

2. Model routing. Use Haiku for planning and search result triage (cheap, fast), switch to Sonnet only for synthesis calls where reasoning quality matters. Split the system prompt accordingly.

3. Page content truncation. The max_chars=4000 parameter in read_page prevents a single verbose page from consuming your entire context budget. For most factual research, 4,000 characters is sufficient.

4. Session-level budget tracking. Log response.usage after every API call and accumulate costs. If cost exceeds your per-session threshold (e.g., $0.50), terminate the loop early.

total_cost = 0.0
SONNET_INPUT = 3.00 / 1_000_000
SONNET_CACHE_READ = 0.30 / 1_000_000
SONNET_OUTPUT = 15.00 / 1_000_000

def track_cost(usage) -> float:
    cost = (
        usage.input_tokens * SONNET_INPUT +
        getattr(usage, "cache_read_input_tokens", 0) * SONNET_CACHE_READ +
        getattr(usage, "cache_creation_input_tokens", 0) * SONNET_INPUT * 1.25 +
        usage.output_tokens * SONNET_OUTPUT
    )
    return cost

FAQ

What search API should I use? Brave Search API ($3/1,000 queries) and Tavily ($1/1,000 queries with AI-optimized extraction) are the best options as of April 2026. Avoid Google Custom Search — the free tier is too limited and the paid tier expensive. For internal enterprise research, skip search entirely and use read_page against known document URLs.

How do I handle paywalled articles? If read_page returns a login wall, skip the URL and use a different source. For enterprise use cases, integrate with a content API (e.g., Diffbot) that handles extraction. Don't attempt to circumvent paywalls.

Can I run searches in parallel? Yes — Claude can issue multiple web_search tool calls in a single response if you configure tool_choice: {"type": "any"} and process all tool results before the next turn. This cuts wall-clock time by 3–5× for the search phase with no change to cost.

How accurate are the reports? Accuracy depends almost entirely on source quality and your search queries. The agent faithfully reports what sources say but cannot verify primary facts independently. Always include source links in reports so readers can verify claims. Mark low-confidence facts explicitly.

Can this agent write to a database or send emails? Yes — add tools for those actions. The pattern is identical: define the schema, implement the function in execute_tool, and add instructions to the system prompt for when to use each tool.

Sources

• Anthropic Tool Use Documentation
• Claude Agent SDK — Anthropic
• Brave Search API Documentation
• Tavily Search API
• BeautifulSoup Documentation
• Related: Claude Agent SDK Guide — agentic loop patterns, tool execution, and production deployment

→ Get Agent SDK Cookbook — $49

The cookbook includes a production-ready research agent with Brave/Tavily integration, cost tracking, async parallel search, and export to Markdown, Notion, and Google Docs — ready to run in under 10 minutes.