Anthropic Claude API Guide: Building Production AI Applications

Python Setup

pip install anthropic

from anthropic import Anthropic

client = Anthropic()  # Uses ANTHROPIC_API_KEY env var

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello, Claude!"}
    ]
)

print(message.content[0].text)

TypeScript/JavaScript Setup

npm install @anthropic-ai/sdk

import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic();

const message = await anthropic.messages.create({
  model: "claude-3-5-sonnet-20241022",
  max_tokens: 1024,
  messages: [
    { role: "user", content: "Hello, Claude!" }
  ]
});

console.log(message.content[0].text);

Analysing Long Documents

from anthropic import Anthropic

client = Anthropic()

# Load your long document
with open("contract.txt", "r") as f:
    document = f.read()

# Analyse in single call - no chunking needed
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=4096,
    messages=[
        {
            "role": "user",
            "content": f"""Analyse this contract and provide:
1. Key terms and conditions
2. Unusual or concerning clauses
3. Missing standard provisions
4. Risk summary with severity ratings

Contract:
{document}"""
        }
    ]
)

print(message.content[0].text)

Compare Multiple Documents

# Compare multiple documents in one call
documents = {
    "proposal_a": load_document("proposal_a.pdf"),
    "proposal_b": load_document("proposal_b.pdf"),
    "proposal_c": load_document("proposal_c.pdf"),
}

prompt = f"""Compare these three vendor proposals for our CRM implementation:

PROPOSAL A:
{documents['proposal_a']}

PROPOSAL B:
{documents['proposal_b']}

PROPOSAL C:
{documents['proposal_c']}

Provide:
1. Comparison table of key features, pricing, timeline
2. Strengths and weaknesses of each
3. Risk analysis
4. Recommendation with justification"""

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=4096,
    messages=[{"role": "user", "content": prompt}]
)

Tool Definition Structure

tools = [
    {
        "name": "get_customer_data",
        "description": "Retrieve customer information from the CRM system. Use this when the user asks about a specific customer.",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "The unique customer identifier (e.g., 'CUS-12345')"
                },
                "include_orders": {
                    "type": "boolean",
                    "description": "Whether to include order history"
                }
            },
            "required": ["customer_id"]
        }
    },
    {
        "name": "search_products",
        "description": "Search the product catalog. Use when the user wants to find or compare products.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search terms"
                },
                "category": {
                    "type": "string",
                    "description": "Product category filter"
                },
                "max_results": {
                    "type": "integer",
                    "description": "Maximum results to return"
                }
            },
            "required": ["query"]
        }
    }
]

Full Tool Use Implementation

from anthropic import Anthropic
import json

client = Anthropic()

def process_tool_call(tool_name, tool_input):
    """Execute the actual tool and return results."""
    if tool_name == "get_customer_data":
        # Your actual implementation
        return get_customer_from_crm(tool_input["customer_id"])
    elif tool_name == "search_products":
        return search_product_catalog(**tool_input)
    else:
        return {"error": f"Unknown tool: {tool_name}"}

def chat_with_tools(user_message):
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # Check if Claude wants to use a tool
        if response.stop_reason == "tool_use":
            # Process each tool call
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = process_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })

            # Add assistant response and tool results to messages
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

        else:
            # No more tool calls, return final response
            return response.content[0].text

# Usage
result = chat_with_tools("What's the order history for customer CUS-12345?")
print(result)

Processing Images with Claude

import base64
from anthropic import Anthropic

client = Anthropic()

# Load image as base64
with open("invoice.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": image_data
                    }
                },
                {
                    "type": "text",
                    "text": "Extract the following from this invoice: vendor name, invoice number, date, line items with amounts, total, and GST amount. Return as JSON."
                }
            ]
        }
    ]
)

print(message.content[0].text)

Compare Multiple Images

# Compare before/after images
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=2048,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Compare these two property photos and describe the changes:"},
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/jpeg", "data": before_image}
                },
                {"type": "text", "text": "BEFORE"},
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/jpeg", "data": after_image}
                },
                {"type": "text", "text": "AFTER"},
                {"type": "text", "text": "List all visible differences and assess the renovation quality."}
            ]
        }
    ]
)

Python Streaming

from anthropic import Anthropic

client = Anthropic()

# Stream response
with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain AI in 200 words"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # Newline at end

TypeScript Streaming (Next.js API Route)

// app/api/chat/route.ts
import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic();

export async function POST(req: Request) {
  const { messages } = await req.json();

  const stream = anthropic.messages.stream({
    model: 'claude-3-5-sonnet-20241022',
    max_tokens: 1024,
    messages
  });

  // Return as Server-Sent Events
  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const event of stream) {
        if (event.type === 'content_block_delta' &&
            event.delta.type === 'text_delta') {
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({text: event.delta.text})}\n\n`)
          );
        }
      }
      controller.enqueue(encoder.encode('data: [DONE]\n\n'));
      controller.close();
    }
  });

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    }
  });
}

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=messages
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                print(f"Calling tool: {event.content_block.name}")
        elif event.type == "content_block_delta":
            if hasattr(event.delta, "text"):
                print(event.delta.text, end="", flush=True)
            elif hasattr(event.delta, "partial_json"):
                # Tool arguments being streamed
                pass

Using Prompt Caching

# Long system prompt or document that's reused
SYSTEM_CONTEXT = "..." # Your long context

# Mark content for caching
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_CONTEXT,
            "cache_control": {"type": "ephemeral"}  # Enable caching
        }
    ],
    messages=[{"role": "user", "content": user_query}]
)

# Cache hits reduce input token costs by 90%

Robust Error Handling

from anthropic import Anthropic, RateLimitError, APIStatusError
from tenacity import retry, stop_after_attempt, wait_exponential

client = Anthropic()

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    retry=lambda e: isinstance(e, (RateLimitError, APIStatusError))
)
def call_claude(messages, model="claude-3-5-sonnet-20241022"):
    try:
        response = client.messages.create(
            model=model,
            max_tokens=1024,
            messages=messages
        )
        return response.content[0].text

    except RateLimitError:
        print("Rate limited, will retry...")
        raise  # Let tenacity retry

    except APIStatusError as e:
        if "overloaded" in str(e):
            print("API overloaded, will retry...")
            raise
        raise  # Other API errors, don't retry

    except Exception as e:
        print(f"Unexpected error: {e}")
        raise