Claude Agent SDK Guide: Build Production AI Agents

from anthropic import Anthropic
from anthropic.lib.agents import Agent, tool

client = Anthropic()

# Define a tool using the @tool decorator
@tool
def search_documents(query: str, top_k: int = 5) -> list[dict]:
    """
    Search the internal document store for relevant documents.

    Args:
        query: Natural language search query
        top_k: Number of results to return (default 5, max 20)
    Returns:
        List of document objects with title, url, and excerpt
    """
    return document_store.search(query, limit=top_k)

@tool
def create_ticket(title: str, description: str, priority: str = "medium") -> dict:
    """
    Create a new support ticket in the ticketing system.

    Args:
        title: Short ticket title (max 120 chars)
        description: Full description of the issue
        priority: One of 'low', 'medium', 'high', 'critical'
    Returns:
        Created ticket with id, url, and status
    """
    return ticketing_system.create(title=title, body=description, priority=priority)

# Create the agent
support_agent = Agent(
    client=client,
    model="claude-sonnet-4-6",
    system="""You are a support desk agent. Your job is to:
    1. Search the documentation to find answers to user questions
    2. Create support tickets for issues that require engineering attention
    3. Always search before creating a ticket — most questions are already answered

    Tone: professional, concise. Never speculate — only report what the docs say.""",
    tools=[search_documents, create_ticket],
    max_tokens=2048,
    max_iterations=15
)

# Synchronous run
result = support_agent.run("A user is getting a 403 error when trying to access /api/reports")

print(result.output)          # Final text response
print(result.tool_calls)      # List of all tool calls made
print(result.usage.total_tokens)  # Token count for the run

# Async run (recommended for production)
import asyncio

async def handle_support_request(user_message: str):
    result = await support_agent.arun(user_message)
    return result.output

# Streaming (for real-time UI feedback)
async def stream_support_response(user_message: str):
    async for event in support_agent.astream(user_message):
        if event.type == "text_delta":
            yield event.delta
        elif event.type == "tool_use_start":
            yield f"\n[Searching: {event.tool_name}...]\n"

from anthropic.lib.agents import Agent, tool, SubAgentRunner

# Define specialist sub-agents
research_agent = Agent(
    client=client,
    model="claude-sonnet-4-6",
    system="You are a research specialist. Search for and summarise information accurately.",
    tools=[web_search, read_document, extract_data],
    max_iterations=10
)

writer_agent = Agent(
    client=client,
    model="claude-sonnet-4-6",
    system="You are a professional business writer. Turn research into clear, compelling reports.",
    tools=[format_document, check_grammar],
    max_iterations=8
)

validator_agent = Agent(
    client=client,
    model="claude-opus-4-6",  # Opus for quality validation
    system="You are a quality validator. Check documents for accuracy, completeness and logical consistency.",
    tools=[verify_facts, check_sources],
    max_iterations=5
)

# Create orchestrator with sub-agent tools
runner = SubAgentRunner(client=client)

@tool
def delegate_research(topic: str, depth: str = "comprehensive") -> str:
    """Run the research specialist agent on a given topic."""
    result = runner.run(research_agent, f"Research {depth}ly: {topic}")
    return result.output

@tool
def delegate_writing(research_notes: str, format: str = "executive brief") -> str:
    """Run the writing specialist to produce a document from research notes."""
    result = runner.run(writer_agent, f"Write a {format} based on:\n{research_notes}")
    return result.output

@tool
def delegate_validation(document: str) -> str:
    """Run the validator agent to quality-check a document."""
    result = runner.run(validator_agent, f"Validate this document:\n{document}")
    return result.output

orchestrator = Agent(
    client=client,
    model="claude-opus-4-6",
    system="""You are an orchestrator. For research tasks:
    1. First delegate research to the research specialist
    2. Then delegate writing to the writing specialist
    3. Finally, delegate validation to the validator
    4. Synthesise the validated output into a final deliverable.""",
    tools=[delegate_research, delegate_writing, delegate_validation],
    max_iterations=12
)

@tool
def fetch_customer_data(customer_id: str) -> dict:
    """
    Fetch customer record from CRM by ID.
    Returns customer data or an error object if not found.
    """
    try:
        customer = crm.get_customer(customer_id)
        return {
            "success": True,
            "customer": {
                "id": customer.id,
                "name": customer.name,
                "email": customer.email,
                "tier": customer.tier,
                "account_since": customer.created_at.isoformat()
            }
        }
    except CustomerNotFoundError:
        return {"success": False, "error": "customer_not_found", "customer_id": customer_id}
    except CRMTimeoutError:
        return {"success": False, "error": "crm_timeout", "retry_after": 30}
    except Exception as e:
        logger.error(f"Unexpected CRM error for {customer_id}: {e}")
        return {"success": False, "error": "unexpected_error"}

from anthropic.lib.agents import Agent, Session

# In-memory session (single process, single conversation)
session = Session()

result1 = project_agent.run(
    "Start tracking a new project: Q2 marketing campaign",
    session=session
)

result2 = project_agent.run(
    "Add a task: design new landing page, owner: Sarah, due April 15",
    session=session  # Same session — agent remembers the project context
)

# Persistent session (survives process restarts)
from anthropic.lib.agents import PersistentSession

session = PersistentSession(
    session_id=f"project-{project_id}",
    storage=redis_storage  # Your storage backend
)

# Session is automatically saved after each run
result = project_agent.run("What's the current status of the Q2 campaign?", session=session)