Anthropic Python SDK

Installation

pip install anthropic>=0.25.0

Basic Usage

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

message = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Analyze this tag list and identify patterns."}
    ]
)
print(message.content[0].text)

Model Selection

Model	ID	Best For
Claude Opus 4.6	claude-opus-4-6	Complex analysis, expert reasoning
Claude Sonnet 4.6	claude-sonnet-4-6	Balanced performance/cost
Claude Haiku 4.5	claude-haiku-4-5-20251001	Fast, lightweight tasks

System Prompts

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    system="You are an industrial automation expert specializing in DCS engineering.",
    messages=[
        {"role": "user", "content": "Review this motor tag list for ISA-5.1 compliance."}
    ]
)

Multi-Turn Conversations

def chat(client: anthropic.Anthropic, history: list, user_message: str) -> tuple[str, list]:
    history.append({"role": "user", "content": user_message})

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=history,
    )

    assistant_text = response.content[0].text
    history.append({"role": "assistant", "content": assistant_text})
    return assistant_text, history

Streaming

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Generate a motor PRT template."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# Or get final message after stream
with client.messages.stream(...) as stream:
    message = stream.get_final_message()

Tool Use (Function Calling)

tools = [
    {
        "name": "validate_tag",
        "description": "Validate an ISA-5.1 tag name and return structured info",
        "input_schema": {
            "type": "object",
            "properties": {
                "tag": {"type": "string", "description": "The tag name to validate"},
                "area": {"type": "integer", "description": "Expected area code"},
            },
            "required": ["tag"],
        },
    }
]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Validate tag 11301.FIC.056A for area 11301"}],
)

# Process tool calls
if response.stop_reason == "tool_use":
    for block in response.content:
        if block.type == "tool_use":
            tool_name = block.name
            tool_input = block.input
            result = handle_tool(tool_name, tool_input)

Vision (Image Input)

import base64
from pathlib import Path

def encode_image(path: str) -> str:
    return base64.standard_b64encode(Path(path).read_bytes()).decode("utf-8")

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": encode_image("p&id_diagram.png"),
                    },
                },
                {"type": "text", "text": "Identify all motor symbols and extract their tag names."},
            ],
        }
    ],
)

Error Handling

from anthropic import APIError, APIConnectionError, RateLimitError, APIStatusError

def safe_claude_call(client: anthropic.Anthropic, prompt: str) -> str | None:
    try:
        message = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        return message.content[0].text

    except RateLimitError:
        # Exponential backoff
        import time
        time.sleep(60)
        return None

    except APIConnectionError as e:
        print(f"Connection error: {e}")
        return None

    except APIStatusError as e:
        print(f"API error {e.status_code}: {e.message}")
        return None

Async Client

import asyncio
import anthropic

async def analyze_batch(prompts: list[str]) -> list[str]:
    client = anthropic.AsyncAnthropic()

    async def call(prompt: str) -> str:
        msg = await client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}],
        )
        return msg.content[0].text

    return await asyncio.gather(*[call(p) for p in prompts])

Usage Tracking

response = client.messages.create(...)

print(response.usage.input_tokens)   # tokens sent
print(response.usage.output_tokens)  # tokens received
# Total cost = input_tokens * price_in + output_tokens * price_out

Integration with Streamlit

import streamlit as st
import anthropic

@st.cache_resource
def get_anthropic_client() -> anthropic.Anthropic:
    return anthropic.Anthropic(api_key=st.secrets["anthropic"]["api_key"])

def stream_to_streamlit(prompt: str) -> str:
    client = get_anthropic_client()
    response_placeholder = st.empty()
    full_text = ""

    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            full_text += text
            response_placeholder.markdown(full_text + "▌")

    response_placeholder.markdown(full_text)
    return full_text

Best Practices

Practice	Why
Use @st.cache_resource for client	Avoid creating new client per request
Store API key in secrets.toml / env	Never hardcode keys
Set max_tokens explicitly	Avoid runaway costs
Use Haiku for classification/routing	10x cheaper than Sonnet
Use Opus for complex analysis	Best reasoning quality
Stream long responses	Better UX, fail faster
Handle RateLimitError with backoff	API has rate limits
Track usage per request	Cost monitoring