Anthropic Claude API

import anthropic
from typing import Optional

# Initialize client
client = anthropic.Anthropic()  # Uses ANTHROPIC_API_KEY env var

# Or with explicit key
client = anthropic.Anthropic(api_key="your-api-key")

from anthropic import Anthropic


def chat_with_claude(
    prompt: str,
    model: str = "claude-sonnet-4-20250514",
    max_tokens: int = 1024
) -> str:
    """Send a message to Claude and get a response."""
    client = Anthropic()
    
    message = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
    
    return message.content[0].text


# Usage
response = chat_with_claude("Explain quantum entanglement in simple terms.")
print(response)

from anthropic import Anthropic
from dataclasses import dataclass, field


@dataclass
class Conversation:
    """Manage multi-turn conversations with Claude."""
    
    client: Anthropic = field(default_factory=Anthropic)
    model: str = "claude-sonnet-4-20250514"
    system: str = ""
    messages: list = field(default_factory=list)
    max_tokens: int = 1024
    
    def add_user_message(self, content: str):
        """Add a user message to the conversation."""
        self.messages.append({"role": "user", "content": content})
    
    def add_assistant_message(self, content: str):
        """Add an assistant message to the conversation."""
        self.messages.append({"role": "assistant", "content": content})
    
    def send(self, user_message: str) -> str:
        """Send a message and get a response."""
        self.add_user_message(user_message)
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            system=self.system,
            messages=self.messages
        )
        
        assistant_message = response.content[0].text
        self.add_assistant_message(assistant_message)
        
        return assistant_message
    
    def reset(self):
        """Clear conversation history."""
        self.messages = []


# Usage
conversation = Conversation(
    system="You are a helpful Python tutor. Explain concepts with examples."
)

response1 = conversation.send("What are decorators?")
print(f"Claude: {response1}\n")

response2 = conversation.send("Can you show a practical example?")
print(f"Claude: {response2}")

from anthropic import Anthropic


class ClaudeAssistant:
    """Claude assistant with configurable personas."""
    
    PERSONAS = {
        "code_reviewer": """You are an expert code reviewer. Your responsibilities:
- Identify bugs, security issues, and performance problems
- Suggest improvements following best practices
- Explain your reasoning clearly
- Rate code quality on a 1-10 scale
- Be constructive and educational in feedback

Format reviews with sections: Summary, Issues, Suggestions, Rating""",

        "technical_writer": """You are a technical documentation expert. Your style:
- Write clear, concise documentation
- Include practical code examples
- Structure content with headers and lists
- Anticipate common questions
- Define technical terms on first use

Always target an audience of intermediate developers.""",

        "sql_expert": """You are a SQL and database optimization expert. You:
- Write efficient, readable SQL queries
- Explain query plans and optimization strategies
- Follow SQL style best practices
- Consider indexing and performance implications
- Support PostgreSQL, MySQL, and SQLite syntax

Always explain your queries and suggest indexes when relevant.""",
    }
    
    def __init__(self, persona: str = None, custom_system: str = None):
        self.client = Anthropic()
        self.system = custom_system or self.PERSONAS.get(persona, "")
    
    def ask(self, prompt: str, model: str = "claude-sonnet-4-20250514") -> str:
        """Send a prompt with the configured system context."""
        response = self.client.messages.create(
            model=model,
            max_tokens=2048,
            system=self.system,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.content[0].text


# Usage
reviewer = ClaudeAssistant(persona="code_reviewer")

code = """
def get_user(id):
    query = f"SELECT * FROM users WHERE id = {id}"
    return db.execute(query)
"""

review = reviewer.ask(f"Review this code:\n\n```python\n{code}\n```")
print(review)

from datetime import datetime
from anthropic import Anthropic


def create_dynamic_system_prompt(
    user_name: str,
    user_role: str,
    context: dict = None
) -> str:
    """Generate a dynamic system prompt based on context."""
    context = context or {}
    
    base_prompt = f"""You are an AI assistant helping {user_name}, who is a {user_role}.

Current date and time: {datetime.now().strftime("%Y-%m-%d %H:%M")}
"""
    
    # Add context-specific instructions
    if context.get("project_type"):
        base_prompt += f"\nProject context: {context['project_type']}"
    
    if context.get("tech_stack"):
        stack = ", ".join(context["tech_stack"])
        base_prompt += f"\nTech stack in use: {stack}"
    
    if context.get("constraints"):
        base_prompt += f"\nConstraints to consider: {context['constraints']}"
    
    base_prompt += """

Adapt your responses to the user's expertise level. Be direct and actionable.
When suggesting code, match the project's existing patterns and stack."""
    
    return base_prompt


# Usage
client = Anthropic()

system = create_dynamic_system_prompt(
    user_name="Alex",
    user_role="Senior Backend Engineer",
    context={
        "project_type": "E-commerce API",
        "tech_stack": ["Python", "FastAPI", "PostgreSQL", "Redis"],
        "constraints": "Must support 10k requests/second"
    }
)

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=system,
    messages=[{"role": "user", "content": "How should I implement rate limiting?"}]
)

print(response.content[0].text)

from anthropic import Anthropic


def stream_response(prompt: str, system: str = "") -> str:
    """Stream a response from Claude."""
    client = Anthropic()
    
    full_response = ""
    
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_response += text
    
    print()  # Newline at end
    return full_response


# Usage
response = stream_response(
    "Write a haiku about programming",
    system="You are a creative poet."
)

import asyncio
from anthropic import AsyncAnthropic


async def async_stream_response(prompt: str) -> str:
    """Async streaming from Claude."""
    client = AsyncAnthropic()
    
    full_response = ""
    
    async with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
            full_response += text
    
    print()
    return full_response


# Usage
asyncio.run(async_stream_response("Explain async programming."))

from anthropic import Anthropic
import json


def get_weather(location: str, unit: str = "celsius") -> dict:
    """Simulated weather API."""
    return {
        "location": location,
        "temperature": 22 if unit == "celsius" else 72,
        "unit": unit,
        "conditions": "sunny"
    }


def search_database(query: str, limit: int = 10) -> list:
    """Simulated database search."""
    return [
        {"id": 1, "title": f"Result for: {query}", "score": 0.95},
        {"id": 2, "title": f"Another result for: {query}", "score": 0.87},
    ]


# Define tools for Claude
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country, e.g., 'London, UK'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        }
    },
    {
        "name": "search_database",
        "description": "Search the knowledge database",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search query"
                },
                "limit": {
                    "type": "integer",
                    "description": "Maximum results to return"
                }
            },
            "required": ["query"]
        }
    }
]

# Tool execution mapping
tool_functions = {
    "get_weather": get_weather,
    "search_database": search_database
}


def run_with_tools(user_message: str) -> str:
    """Run a conversation with tool use."""
    client = Anthropic()
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        
        # Check if Claude wants to use tools
        if response.stop_reason == "tool_use":
            # Process tool calls
            tool_results = []
            
            for block in response.content:
                if block.type == "tool_use":
                    tool_name = block.name
                    tool_input = block.input
                    
                    # Execute the tool
                    if tool_name in tool_functions:
                        result = tool_functions[tool_name](**tool_input)
                    else:
                        result = {"error": f"Unknown tool: {tool_name}"}
                    
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
            
            # Add assistant message and tool results
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        
        else:
            # No more tool calls, return final response
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            
            return ""


# Usage
response = run_with_tools("What's the weather in Tokyo and Paris?")
print(response)

import base64
import httpx
from anthropic import Anthropic
from pathlib import Path


def encode_image_to_base64(image_path: str) -> str:
    """Encode a local image to base64."""
    with open(image_path, "rb") as f:
        return base64.standard_b64encode(f.read()).decode("utf-8")


def get_image_media_type(image_path: str) -> str:
    """Get media type from file extension."""
    suffix = Path(image_path).suffix.lower()
    media_types = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }
    return media_types.get(suffix, "image/jpeg")


def analyze_image(image_source: str, prompt: str) -> str:
    """Analyze an image with Claude Vision."""
    client = Anthropic()
    
    # Determine if URL or file path
    if image_source.startswith(("http://", "https://")):
        # Fetch and encode URL
        image_data = base64.standard_b64encode(
            httpx.get(image_source).content
        ).decode("utf-8")
        media_type = "image/jpeg"  # Default for URLs
    else:
        # Local file
        image_data = encode_image_to_base64(image_source)
        media_type = get_image_media_type(image_source)
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data
                        }
                    },
                    {
                        "type": "text",
                        "text": prompt
                    }
                ]
            }
        ]
    )
    
    return response.content[0].text


def compare_images(image_paths: list[str], prompt: str) -> str:
    """Compare multiple images with Claude."""
    client = Anthropic()
    
    content = []
    
    for path in image_paths:
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": get_image_media_type(path),
                "data": encode_image_to_base64(path)
            }
        })
    
    content.append({"type": "text", "text": prompt})
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": content}]
    )
    
    return response.content[0].text


# Usage
analysis = analyze_image(
    "screenshot.png",
    "Describe this UI and suggest improvements."
)
print(analysis)

from anthropic import Anthropic


def solve_with_thinking(problem: str, budget_tokens: int = 10000) -> dict:
    """Use Claude's extended thinking for complex problems."""
    client = Anthropic()
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=16000,
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens
        },
        messages=[{"role": "user", "content": problem}]
    )
    
    result = {
        "thinking": "",
        "answer": ""
    }
    
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["answer"] = block.text
    
    return result


# Usage
problem = """
A farmer has 100 meters of fencing and wants to enclose a rectangular 
area next to a river (no fencing needed on the river side). What 
dimensions maximize the enclosed area?
"""

solution = solve_with_thinking(problem)
print("Thinking process:")
print(solution["thinking"][:500] + "...")
print("\nFinal answer:")
print(solution["answer"])

from anthropic import Anthropic


def count_tokens(text: str, model: str = "claude-sonnet-4-20250514") -> int:
    """Count tokens in text using Anthropic's tokenizer."""
    client = Anthropic()
    
    # Use the count_tokens endpoint
    result = client.messages.count_tokens(
        model=model,
        messages=[{"role": "user", "content": text}]
    )
    
    return result.input_tokens


class CostTracker:
    """Track Claude API usage and costs."""
    
    # Pricing per million tokens (as of 2024)
    PRICING = {
        "claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
        "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00},
        "claude-3-opus-20240229": {"input": 15.00, "output": 75.00},
        "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
    }
    
    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.requests = []
    
    def record_usage(self, model: str, input_tokens: int, output_tokens: int):
        """Record token usage from a request."""
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (
            (input_tokens / 1_000_000) * pricing["input"] +
            (output_tokens / 1_000_000) * pricing["output"]
        )
        
        self.requests.append({
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        })
    
    def get_total_cost(self) -> float:
        """Get total cost across all requests."""
        return sum(r["cost"] for r in self.requests)
    
    def get_summary(self) -> dict:
        """Get usage summary."""
        return {
            "total_requests": len(self.requests),
            "total_input_tokens": self.total_input_tokens,
            "total_output_tokens": self.total_output_tokens,
            "total_cost": self.get_total_cost()
        }


# Usage with tracking
tracker = CostTracker()
client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

tracker.record_usage(
    "claude-sonnet-4-20250514",
    response.usage.input_tokens,
    response.usage.output_tokens
)

print(tracker.get_summary())

import time
from anthropic import Anthropic, APIError, RateLimitError, APIConnectionError


def robust_claude_call(
    messages: list,
    model: str = "claude-sonnet-4-20250514",
    max_retries: int = 3,
    base_delay: float = 1.0
) -> str:
    """Make a robust Claude API call with retries."""
    client = Anthropic()
    
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model=model,
                max_tokens=1024,
                messages=messages
            )
            return response.content[0].text
            
        except RateLimitError as e:
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                print(f"Rate limited. Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise
                
        except APIConnectionError as e:
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                print(f"Connection error. Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise
                
        except APIError as e:
            # Don't retry on client errors (4xx)
            if e.status_code and 400 <= e.status_code < 500:
                raise
            
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                print(f"API error. Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise
    
    raise RuntimeError("Max retries exceeded")


# Usage
response = robust_claude_call([
    {"role": "user", "content": "Hello!"}
])