Anthropic Claude Agent SDK Integration for Agent Building

Anthropic Claude Agent SDK Integration

Claude Agent SDK (claude-agent-sdk) is Anthropic's official Python SDK for building agents based on Claude. Provides high-level abstractions over the Anthropic API: agent lifecycle management, tool integration, Model Context Protocol (MCP) support, event streaming, and built-in human-in-the-loop.

Installation and Basic Agent

# pip install anthropic claude-agent-sdk
import anthropic
from claude_agent_sdk import Agent, AgentConfig, tool

client = anthropic.Anthropic()

# Tool definition via decorator
@tool
def search_database(query: str, table: str = "products") -> str:
    """Search company database.

    Args:
        query: Search query
        table: Table to search (products, orders, customers)
    """
    results = db.search(query=query, table=table, limit=10)
    return results.to_json()

@tool
def create_support_ticket(
    customer_id: str,
    subject: str,
    description: str,
    priority: str = "normal",
) -> str:
    """Create support ticket.

    Args:
        customer_id: Customer ID
        subject: Ticket subject
        description: Detailed description
        priority: Priority level (low, normal, high, critical)
    """
    ticket = helpdesk.create_ticket(
        customer_id=customer_id,
        subject=subject,
        description=description,
        priority=priority,
    )
    return f"Ticket #{ticket['id']} created. URL: {ticket['url']}"

# Agent configuration and creation
config = AgentConfig(
    model="claude-opus-4-5",
    system_prompt="""You are a customer support agent for TechCorp.
Help customers solve problems using available tools.
Always verify data through tools — don't rely on memory.""",
    max_turns=10,
)

agent = Agent(
    client=client,
    config=config,
    tools=[search_database, create_support_ticket],
)

# Run agent
result = agent.run(
    messages=[{"role": "user", "content": "Customer ID 12345 has an issue with order #99876"}]
)
print(result.final_message)

Agent Event Streaming

import asyncio

async def run_agent_with_streaming():
    async for event in agent.astream(
        messages=[{"role": "user", "content": "Analyze last 10 orders for customer ID 12345"}]
    ):
        match event.type:
            case "text_delta":
                print(event.text, end="", flush=True)
            case "tool_use_start":
                print(f"\n[Tool: {event.tool_name}]")
            case "tool_result":
                print(f"[Result received, {len(event.content)} characters]")
            case "agent_turn_complete":
                print(f"\n[Completed in {event.turn_count} turns]")

asyncio.run(run_agent_with_streaming())

MCP (Model Context Protocol) Integration

from claude_agent_sdk import Agent, MCPServerConfig

# MCP servers extend the agent with tools without explicit definition
agent_with_mcp = Agent(
    client=client,
    config=config,
    mcp_servers=[
        MCPServerConfig(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],
        ),
        MCPServerConfig(
            name="postgres",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"POSTGRES_URL": "postgresql://user:pass@localhost/db"},
        ),
        MCPServerConfig(
            name="github",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
            env={"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."},
        ),
    ],
)

# Agent can now work with files, PostgreSQL, and GitHub
result = agent_with_mcp.run(
    messages=[{"role": "user", "content": "Read config.yaml and create GitHub Issues based on TODO comments"}]
)

Multi-turn Dialog with History

from claude_agent_sdk import ConversationSession

# Session preserves dialog history
session = ConversationSession(
    agent=agent,
    session_id="customer_session_12345",
)

# First turn
response1 = session.send("What is the status of my order #99876?")

# Second turn — agent remembers context
response2 = session.send("Can I reschedule delivery for tomorrow?")

# Third turn
response3 = session.send("Please confirm")

print(session.get_history())

Human-in-the-loop via Approval

from claude_agent_sdk import Agent, ToolApprovalPolicy

class CustomApprovalPolicy(ToolApprovalPolicy):
    """Requests confirmation for destructive operations"""

    REQUIRES_APPROVAL = {"delete_order", "process_refund", "ban_customer"}

    async def should_approve(self, tool_name: str, tool_input: dict) -> bool:
        if tool_name not in self.REQUIRES_APPROVAL:
            return True  # Auto-approve safe tools

        # Notify operator
        await notify_operator(
            message=f"Approval required: {tool_name}\nParameters: {tool_input}",
            callback_url="/api/approve/{approval_id}",
        )

        # Wait for decision (5-minute timeout)
        approval = await wait_for_approval(timeout=300)
        return approval.approved

agent_with_approval = Agent(
    client=client,
    config=config,
    tools=[search_database, process_refund, ban_customer],
    approval_policy=CustomApprovalPolicy(),
)

Parallel Agents

import asyncio

async def run_parallel_analysis(customer_ids: list[str]):
    """Parallel analysis of multiple customers"""

    async def analyze_customer(customer_id: str):
        return await agent.arun(
            messages=[{
                "role": "user",
                "content": f"Analyze activity for customer {customer_id} over the past month"
            }]
        )

    results = await asyncio.gather(*[
        analyze_customer(cid) for cid in customer_ids
    ], return_exceptions=True)

    return {
        cid: result.final_message if not isinstance(result, Exception) else str(result)
        for cid, result in zip(customer_ids, results)
    }

Practical Case Study: Financial Monitoring Agent

Task: automatic transaction monitoring to detect suspicious activity. Financial officer spent 3 hours daily manually reviewing rule-based system flags.

Agent Tools:

• get_flagged_transactions — flags list for period
• get_transaction_history — customer transaction history
• get_customer_profile — KYC profile
• check_external_sanctions — sanctions list verification
• create_sar_draft — Suspicious Activity Report draft
• escalate_to_officer — escalation for financial officer

Workflow:

1. Agent receives 50–200 flags daily
2. For each flag: analyzes context, history, profile
3. Makes determination: false positive / suspicious / critical
4. For critical cases: creates SAR draft and escalates
5. Auto-closes false positives

Results:

• False positives auto-processed: 78%
• Financial officer time: 3 hours → 45 minutes (only real cases)
• SAR draft quality (compliance officer rating): 4.3/5.0
• Critical case response time: 4–8 hours → 15 minutes

Timeline

• Basic agent with tools: 3–5 days
• MCP integrations: 1–3 days per server
• Human-in-the-loop with approval flow: 1 week
• Production deployment with monitoring: 1 week