# Anthropic Python SDK

The Anthropic Python SDK provides convenient access to the Anthropic REST API for building applications powered by Claude AI models. The library offers type definitions for all request parameters and response fields, supports both synchronous and asynchronous clients powered by httpx, and works with Python 3.9+. It includes comprehensive support for the Messages API, streaming responses, tool use (function calling), token counting, message batches, and integration with AWS Bedrock and Google Vertex AI platforms.

The SDK follows a resource-based architecture where the main client exposes different API resources like `messages`, `models`, and `beta`. Each resource provides methods corresponding to API endpoints, with full type safety through Pydantic models and TypedDict parameters. The library automatically handles authentication, retries, timeouts, and error handling, making it straightforward to integrate Claude's capabilities into any Python application.

## Creating a Message

The `messages.create()` method sends a conversation to Claude and receives a response. This is the primary way to interact with Claude models for text generation, including multi-turn conversations.

```python
import os
from anthropic import Anthropic

client = Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),  # This is the default and can be omitted
)

# Basic message creation
message = client.messages.create(
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello, Claude"}
    ],
    model="claude-sonnet-4-5-20250929",
)
print(message.content)
# Output: [TextBlock(text='Hello! How can I help you today?', type='text')]

# Multi-turn conversation
response = client.messages.create(
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello!"},
        {"role": "assistant", "content": "Hello! How can I help you today?"},
        {"role": "user", "content": "What is the capital of France?"},
    ],
    model="claude-sonnet-4-5-20250929",
)
print(response.content[0].text)
# Output: The capital of France is Paris.
```

## Async Message Creation

The `AsyncAnthropic` client provides the same interface as the synchronous client but with async/await support for non-blocking operations in asyncio applications.

```python
import os
import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
)

async def main() -> None:
    message = await client.messages.create(
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Hello, Claude"}
        ],
        model="claude-sonnet-4-5-20250929",
    )
    print(message.content)
    # Output: [TextBlock(text='Hello! How can I help you today?', type='text')]

asyncio.run(main())
```

## Streaming Responses with stream=True

Enable streaming by setting `stream=True` to receive Server-Sent Events (SSE) as the response is generated, allowing real-time display of output.

```python
from anthropic import Anthropic

client = Anthropic()

# Basic streaming with stream=True
stream = client.messages.create(
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Tell me a short story about a robot."}
    ],
    model="claude-sonnet-4-5-20250929",
    stream=True,
)

for event in stream:
    if event.type == "content_block_delta":
        if event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
print()
# Output: (streams text incrementally as it's generated)
```

## Streaming Helpers with messages.stream()

The `messages.stream()` context manager provides higher-level streaming helpers including text accumulation, event processing, and convenient access to the final message.

```python
import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Say hello there!"}
        ],
        model="claude-sonnet-4-5-20250929",
    ) as stream:
        # Stream text as it arrives
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        print()

        # Or iterate over events
        # async for event in stream:
        #     if event.type == "text":
        #         print(event.text, end="", flush=True)

    # Get the accumulated final message after stream completes
    message = await stream.get_final_message()
    print(f"Total tokens: {message.usage.input_tokens + message.usage.output_tokens}")

asyncio.run(main())
```

## Tool Use (Function Calling)

Define tools that Claude can use during a conversation. When Claude decides to use a tool, you receive a tool_use block that you can process and return results for.

```python
from anthropic import Anthropic
from anthropic.types import ToolParam, MessageParam

client = Anthropic()

# Define a tool with JSON schema
tools: list[ToolParam] = [
    {
        "name": "get_weather",
        "description": "Get the weather for a specific location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City and state"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        },
    }
]

# Initial request
message = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is the weather in San Francisco?"}],
    tools=tools,
)

# Check if Claude wants to use a tool
if message.stop_reason == "tool_use":
    tool_use = next(c for c in message.content if c.type == "tool_use")
    print(f"Tool called: {tool_use.name}")
    print(f"Input: {tool_use.input}")
    # Output: Tool called: get_weather
    # Output: Input: {'location': 'San Francisco'}

    # Send tool result back
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "What is the weather in San Francisco?"},
            {"role": "assistant", "content": message.content},
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": "The weather is 72°F and sunny."
                    }
                ],
            },
        ],
        tools=tools,
    )
    print(response.content[0].text)
    # Output: The weather in San Francisco is currently 72°F and sunny.
```

## Tool Runner with @beta_tool Decorator

The `@beta_tool` decorator and `tool_runner()` method provide a convenient way to define tools as Python functions and have them automatically called by the SDK.

```python
import json
from typing_extensions import Literal
from anthropic import Anthropic, beta_tool

client = Anthropic()

@beta_tool
def get_weather(location: str, units: Literal["c", "f"] = "f") -> str:
    """Lookup the weather for a given city

    Args:
        location: The city and state, e.g. San Francisco, CA
        units: Unit for the output, either 'c' for celsius or 'f' for fahrenheit
    Returns:
        A JSON string with weather information.
    """
    # Simulated weather API response
    return json.dumps({
        "location": location,
        "temperature": "20°C" if units == "c" else "68°F",
        "condition": "Sunny",
    })

# The tool_runner automatically calls tools and continues the conversation
runner = client.beta.messages.tool_runner(
    max_tokens=1024,
    model="claude-sonnet-4-5-20250929",
    tools=[get_weather],
    messages=[{"role": "user", "content": "What is the weather in SF?"}],
)

for message in runner:
    print(f"Message: {message.content}")
    # Output: First message shows tool_use block
    # Output: Second message shows final text response with weather info
```

## Structured Outputs with Pydantic

The `messages.parse()` method extracts structured data from responses using Pydantic models for type-safe output parsing.

```python
import pydantic
import anthropic

class Order(pydantic.BaseModel):
    product_name: str
    price: float
    quantity: int

client = anthropic.Anthropic()

prompt = """
Extract the product name, price, and quantity from this customer message:
"Hi, I'd like to order 2 packs of Green Tea for 5.50 dollars each."
"""

parsed_message = client.messages.parse(
    model="claude-sonnet-4-5",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=1024,
    output_format=Order,
)

print(parsed_message.parsed_output)
# Output: Order(product_name='Green Tea', price=5.5, quantity=2)
print(f"Total: ${parsed_message.parsed_output.price * parsed_message.parsed_output.quantity}")
# Output: Total: $11.0
```

## Extended Thinking

Enable extended thinking to have Claude show its reasoning process before providing a final answer, useful for complex problems.

```python
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=3200,
    thinking={"type": "enabled", "budget_tokens": 1600},
    messages=[{"role": "user", "content": "Solve this step by step: What is 15% of 340?"}],
)

for block in response.content:
    if block.type == "thinking":
        print(f"Thinking: {block.thinking}")
        # Output: Thinking: To find 15% of 340, I need to...
    elif block.type == "text":
        print(f"Answer: {block.text}")
        # Output: Answer: 15% of 340 is 51.
```

## Token Counting

Count tokens before making requests to estimate costs and ensure content fits within context limits.

```python
from anthropic import Anthropic

client = Anthropic()

# Count tokens for a message
count = client.messages.count_tokens(
    model="claude-sonnet-4-5-20250929",
    messages=[
        {"role": "user", "content": "Hello, world! How are you today?"}
    ]
)
print(f"Input tokens: {count.input_tokens}")
# Output: Input tokens: 14

# Check usage after a request
message = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=100,
    messages=[{"role": "user", "content": "Hi!"}],
)
print(f"Usage: {message.usage}")
# Output: Usage(input_tokens=8, output_tokens=15)
```

## Image Analysis

Send images to Claude for analysis by including them as base64-encoded data or URLs in the message content.

```python
import base64
from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# With base64-encoded image
image_data = base64.standard_b64encode(Path("image.png").read_bytes()).decode("utf-8")

response = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What do you see in this image?"},
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": image_data,
                    },
                },
            ],
        },
    ],
    model="claude-sonnet-4-5-20250929",
)
print(response.content[0].text)
# Output: I see [description of the image content]

# With URL source
response = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this image."},
                {
                    "type": "image",
                    "source": {
                        "type": "url",
                        "url": "https://example.com/image.jpg",
                    },
                },
            ],
        },
    ],
    model="claude-sonnet-4-5-20250929",
)
```

## Web Search Tool

Enable Claude to search the web for current information using the built-in web search tool.

```python
from anthropic import Anthropic

client = Anthropic()

message = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What are the latest news about AI today?"}],
    tools=[
        {
            "name": "web_search",
            "type": "web_search_20250305",
        }
    ],
)

# Extract text response
for content_block in message.content:
    if content_block.type == "text":
        print(content_block.text)

# Check web search usage
if message.usage.server_tool_use:
    print(f"Web searches performed: {message.usage.server_tool_use.web_search_requests}")
```

## Message Batches

Process multiple messages in batch for high-throughput scenarios with reduced cost per request.

```python
from anthropic import Anthropic

client = Anthropic()

# Create a batch of requests
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": "request-1",
            "params": {
                "model": "claude-sonnet-4-5-20250929",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "Hello, world"}],
            },
        },
        {
            "custom_id": "request-2",
            "params": {
                "model": "claude-sonnet-4-5-20250929",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "What is 2+2?"}],
            },
        },
    ]
)
print(f"Batch ID: {batch.id}")
print(f"Status: {batch.processing_status}")

# Check batch status
batch = client.messages.batches.retrieve(batch.id)
print(f"Processing status: {batch.processing_status}")

# Get results when processing is complete
if batch.processing_status == "ended":
    result_stream = client.messages.batches.results(batch.id)
    for entry in result_stream:
        print(f"Custom ID: {entry.custom_id}")
        if entry.result.type == "succeeded":
            print(f"Response: {entry.result.message.content}")
```

## Pagination for List Operations

Use auto-paginating iterators to efficiently iterate through paginated API responses.

```python
from anthropic import Anthropic

client = Anthropic()

# Auto-pagination - fetches pages automatically as needed
all_batches = []
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)
    print(f"Batch: {batch.id}, Status: {batch.processing_status}")

# Manual pagination control
first_page = client.messages.batches.list(limit=20)
print(f"First page has {len(first_page.data)} items")

if first_page.has_next_page():
    next_page = first_page.get_next_page()
    print(f"Next page has {len(next_page.data)} items")
```

## AWS Bedrock Integration

Use Claude through AWS Bedrock with the `AnthropicBedrock` client, which handles AWS authentication automatically.

```python
# pip install anthropic[bedrock]
from anthropic import AnthropicBedrock

# Uses AWS credentials from environment or ~/.aws/credentials
client = AnthropicBedrock(
    # Optional: specify credentials explicitly
    # aws_region="us-east-1",
    # aws_access_key="...",
    # aws_secret_key="...",
)

message = client.messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
    model="anthropic.claude-sonnet-4-5-20250929-v1:0",
)
print(message.content[0].text)

# Streaming also works
with client.messages.stream(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Tell me a story."}],
    model="anthropic.claude-sonnet-4-5-20250929-v1:0",
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
```

## Google Vertex AI Integration

Use Claude through Google Vertex AI with the `AnthropicVertex` client.

```python
# pip install anthropic[vertex]
from anthropic import AnthropicVertex

# Uses Google Cloud credentials from environment
client = AnthropicVertex(
    # Optional: specify project and region
    # project_id="your-project-id",
    # region="us-east5",
)

message = client.messages.create(
    model="claude-sonnet-4@20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)
print(message.content[0].text)
```

## Error Handling

Handle API errors with specific exception types for different error conditions.

```python
import anthropic
from anthropic import Anthropic

client = Anthropic()

try:
    message = client.messages.create(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}],
        model="claude-sonnet-4-5-20250929",
    )
except anthropic.APIConnectionError as e:
    print(f"Connection error: {e.__cause__}")
except anthropic.RateLimitError as e:
    print(f"Rate limited (429): {e.response}")
except anthropic.AuthenticationError as e:
    print(f"Authentication failed (401): {e.message}")
except anthropic.BadRequestError as e:
    print(f"Bad request (400): {e.message}")
except anthropic.APIStatusError as e:
    print(f"API error {e.status_code}: {e.message}")
    print(f"Request ID: {e.response.headers.get('request-id')}")
```

## Retries and Timeouts

Configure automatic retry behavior and request timeouts at the client or per-request level.

```python
import httpx
from anthropic import Anthropic

# Configure client-level defaults
client = Anthropic(
    max_retries=5,  # Default is 2
    timeout=30.0,   # Default is 600 seconds (10 minutes)
)

# Fine-grained timeout control
client = Anthropic(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override per-request
message = client.with_options(
    max_retries=10,
    timeout=120.0,
).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
    model="claude-sonnet-4-5-20250929",
)
```

## File Uploads (Beta)

Upload and manage files for use with Claude's capabilities.

```python
from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# Upload a file
file = client.beta.files.upload(
    file=Path("/path/to/document.pdf"),
)
print(f"File ID: {file.id}")

# List files
for f in client.beta.files.list():
    print(f"File: {f.id}, Name: {f.filename}")

# Get file metadata
metadata = client.beta.files.retrieve_metadata(file.id)
print(f"File size: {metadata.size_bytes}")

# Download file content
content = client.beta.files.download(file.id)

# Delete a file
client.beta.files.delete(file.id)
```

## Raw Response Access

Access raw HTTP response data including headers when needed for debugging or advanced use cases.

```python
from anthropic import Anthropic

client = Anthropic()

# Get raw response with headers
response = client.messages.with_raw_response.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
    model="claude-sonnet-4-5-20250929",
)

print(f"Request ID: {response.headers.get('request-id')}")
print(f"Status: {response.status_code}")

# Parse the response body
message = response.parse()
print(message.content[0].text)

# Or access request_id directly from any response
message = client.messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
    model="claude-sonnet-4-5-20250929",
)
print(f"Request ID: {message._request_id}")
```

## Summary

The Anthropic Python SDK is designed for building production applications with Claude AI. The most common use cases include conversational AI assistants using the Messages API, content generation with streaming for real-time output, structured data extraction with Pydantic models, and intelligent agents using tool use for external integrations. The SDK handles authentication, retries, and errors automatically while providing full type safety through Pydantic models and TypedDict parameters.

For integration patterns, applications typically start with the synchronous `Anthropic` client for simple scripts and migrate to `AsyncAnthropic` for web applications and high-concurrency scenarios. Production deployments often leverage AWS Bedrock or Google Vertex AI for enterprise compliance and scaling, using the dedicated `AnthropicBedrock` or `AnthropicVertex` clients. The streaming helpers with `messages.stream()` are recommended for any user-facing application where response latency matters, while Message Batches provide the most cost-effective approach for offline processing of large volumes of requests.