Running Claude Sonnet 4.6 in Production While Scouting the Move to Sonnet 5

import anthropic
 
client = anthropic.Anthropic()
 
 
def first_text(message) -> str:
    """Return the first text block of a response.
 
    content[0] is not guaranteed to be text. With thinking enabled the
    thinking blocks arrive first, and tool-use replies lead with tool_use.
    Selecting by type instead of by position is the whole function — and it
    is what keeps this code working when you swap in Sonnet 5.
    """
    for block in message.content:
        if block.type == "text":
            return block.text
    return ""
 
 
def analyze_codebase(file_paths: list[str]) -> str:
    """
    Load an entire codebase into context for holistic analysis.
    Works well for repositories up to ~800K tokens (leave buffer).
    """
    files_content = ""
    for path in file_paths:
        with open(path, "r") as f:
            files_content += f"\n\n=== {path} ===\n{f.read()}"
 
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=8192,
        messages=[
            {
                "role": "user",
                "content": f"""Analyze this entire codebase and provide:
1. Architectural issues and improvement opportunities
2. Security vulnerabilities and risk assessment
3. Performance bottlenecks
4. Code quality observations
 
Codebase:
{files_content}
"""
            }
        ]
    )
    return first_text(message)
 
# Expected output: Comprehensive architecture report with specific file/line references

import Anthropic from "@anthropic-ai/sdk";
 
interface Message {
  role: "user" | "assistant";
  content: string;
}
 
class PersistentSession {
  private history: Message[] = [];
  private client: Anthropic;
 
  constructor() {
    this.client = new Anthropic();
  }
 
  async chat(userMessage: string): Promise<string> {
    this.history.push({ role: "user", content: userMessage });
 
    const response = await this.client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 4096,
      system:
        "You are a dedicated engineering assistant with full context of this project's development history.",
      messages: this.history, // Full history — no truncation needed up to ~900K tokens
    });
 
    const textBlock = response.content.find((b) => b.type === "text");
    const reply = textBlock
      ? textBlock.text
      : "";
    this.history.push({ role: "assistant", content: reply });
    return reply;
  }
 
  estimateTokensUsed(): number {
    return this.history.reduce((sum, m) => sum + m.content.length / 4, 0);
  }
}

def extract_from_reports(documents: list[str]) -> str:
    """
    Extract structured data from multiple business reports in a single call.
    Much faster and cheaper than processing documents individually.
    """
    combined = "\n\n".join([
        f"=== Document {i+1} ===\n{doc}"
        for i, doc in enumerate(documents)
    ])
 
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=16384,
        messages=[{
            "role": "user",
            "content": f"""Extract structured data from all documents below.
Return valid JSON with this schema for each document:
{{
  "documents": [
    {{
      "revenue": "string",
      "profit_margin": "string",
      "key_metrics": ["string"],
      "risk_factors": ["string"],
      "summary": "string"
    }}
  ]
}}
 
Documents:
{combined}"""
        }]
    )
    return first_text(response)
# Expected output: Valid JSON with extracted fields from each document

import anthropic
 
client = anthropic.Anthropic()
 
def think_deeply(problem: str, budget_tokens: int = 10000) -> dict:
    """
    Use Extended Thinking for problems that benefit from deep reasoning.
 
    budget_tokens: How many tokens the model can use for internal thinking.
                   Range: 1,024 to 32,000. Higher = more thorough reasoning.
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=16000,  # Must exceed budget_tokens + expected output
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens,
        },
        messages=[{"role": "user", "content": problem}],
    )
 
    result = {"thinking": None, "answer": None}
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking  # Internal reasoning trace
        elif block.type == "text":
            result["answer"] = block.text  # Final response
 
    return result
 
# Example: System architecture decision with trade-off analysis
problem = """
We're migrating a Python Flask monolith (PostgreSQL, 100k daily users)
to microservices. Target: 99.99% SLA, horizontal scaling, team-independent deploys.
What's the optimal migration strategy, and which service should we extract first?
Justify your reasoning with specific risk and benefit analysis.
"""
 
result = think_deeply(problem, budget_tokens=15000)
print("Reasoning trace:", result["thinking"][:300], "...")
print("\nFinal answer:", result["answer"])
# Expected output: Detailed migration strategy with prioritized service extraction plan and risk assessment

def think_deeply_v5(problem: str, effort: str = "high") -> dict:
    """Sonnet 5 version. Depth comes from effort, not from a token budget."""
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=16000,
        thinking={"type": "adaptive", "display": "summarized"},
        output_config={"effort": effort},  # lower effort = shallower, faster
        messages=[{"role": "user", "content": problem}],
        # temperature / top_p / top_k are omitted: non-default values return 400
    )
 
    out = {"thinking": None, "answer": None}
    for block in response.content:
        if block.type == "thinking":
            out["thinking"] = block.thinking
        elif block.type == "text":
            out["answer"] = block.text
    return out
 
# Expected output: same dict shape as think_deeply, with summarized thinking

import anthropic
import base64
 
client = anthropic.Anthropic()
 
def execute_computer_task(task: str, screenshot_b64: str) -> dict:
    """
    Given a screenshot and task description, return the actions Sonnet 4.6
    recommends to accomplish the task.
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        tools=[
            {
                "type": "computer_20251124",
                "name": "computer",
                "display_width_px": 1920,
                "display_height_px": 1080,
            }
        ],
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": screenshot_b64,
                        },
                    },
                    {
                        "type": "text",
                        "text": f"Look at the current screen and complete this task: {task}"
                    }
                ],
            }
        ],
    )
 
    actions = []
    for block in response.content:
        if block.type == "tool_use" and block.name == "computer":
            actions.append({
                "action": block.input.get("action"),
                "coordinate": block.input.get("coordinate"),
                "text": block.input.get("text"),
            })
 
    return {
        "actions": actions,
        "reasoning": next(
            (b.text for b in response.content if b.type == "text"), ""
        ),
    }
 
# Expected output: {"actions": [{"action": "click", "coordinate": [960, 540]}], "reasoning": "..."}

HIGH_RISK_KEYWORDS = ["delete", "submit", "purchase", "send", "transfer"]
 
def is_high_risk(action: dict) -> bool:
    action_str = str(action).lower()
    return any(keyword in action_str for keyword in HIGH_RISK_KEYWORDS)
 
def safe_execute(action: dict) -> bool:
    if is_high_risk(action):
        # In production: send Slack notification, await approval webhook
        print(f"⚠️ High-risk action detected: {action}")
        approval = input("Approve? (yes/no): ")
        return approval.strip().lower() == "yes"
    return True

import anthropic
 
client = anthropic.Anthropic()
 
SYSTEM_CONTEXT = """
You are the AI assistant for Acme Corp.
[Include extensive company knowledge, guidelines, product docs here...]
The more text here, the greater the caching savings per request.
"""
 
def cached_query(user_message: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=[
            {
                "type": "text",
                "text": SYSTEM_CONTEXT,
                "cache_control": {"type": "ephemeral"},  # Enable caching
            }
        ],
        messages=[{"role": "user", "content": user_message}],
    )
 
    # Monitor cache efficiency
    usage = response.usage
    cached_tokens = getattr(usage, "cache_read_input_tokens", 0)
    new_tokens = getattr(usage, "cache_creation_input_tokens", 0)
    print(f"Input: {usage.input_tokens} | Cached reads: {cached_tokens} | New writes: {new_tokens}")
 
    return first_text(response)
 
# First call: Cache write at $3.75/1M tokens
first = cached_query("What are our Q1 priorities?")
 
# Subsequent calls: Cache read at $0.30/1M tokens — 90% cheaper
second = cached_query("Summarize the product roadmap.")

import anthropic
import time
 
client = anthropic.Anthropic()
 
def batch_generate(tasks: list[dict]) -> list[str]:
    """
    Process multiple generation tasks asynchronously.
    Cost: 50% off standard pricing.
    Turnaround: within 24 hours.
    """
    requests = [
        {
            "custom_id": f"task-{i}",
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 8192,
                "messages": [{"role": "user", "content": task["prompt"]}],
            },
        }
        for i, task in enumerate(tasks)
    ]
 
    batch = client.messages.batches.create(requests=requests)
    print(f"Batch {batch.id} created with {len(requests)} requests")
 
    # Poll for completion
    while True:
        status = client.messages.batches.retrieve(batch.id)
        if status.processing_status == "ended":
            break
        print(f"Processing... {status.request_counts.processing} remaining")
        time.sleep(30)
 
    # Collect results
    return [
        first_text(result.result.message)
        for result in client.messages.batches.results(batch.id)
        if result.result.type == "succeeded"
    ]
 
# Example: Generate 100 product descriptions in one batch
tasks = [
    {"prompt": f"Write a compelling 150-word product description for: {product}"}
    for product in product_list
]
descriptions = batch_generate(tasks)

import anthropic
 
client = anthropic.Anthropic()
 
def stream_to_console(prompt: str) -> str:
    """Stream response tokens as they arrive."""
    full_text = ""
 
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_text += text
 
        usage = stream.get_final_message().usage
        print(f"\n\n[Input: {usage.input_tokens} | Output: {usage.output_tokens}]")
 
    return full_text

// Next.js App Router streaming with Vercel AI SDK
// app/api/chat/route.ts
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
export async function POST(req: Request) {
  const { messages } = await req.json();
 
  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    async start(controller) {
      const response = await client.messages.create({
        model: "claude-sonnet-4-6",
        max_tokens: 4096,
        stream: true,
        messages,
      });
 
      for await (const event of response) {
        if (
          event.type === "content_block_delta" &&
          event.delta.type === "text_delta"
        ) {
          controller.enqueue(encoder.encode(event.delta.text));
        }
      }
      controller.close();
    },
  });
 
  return new Response(stream, {
    headers: { "Content-Type": "text/plain; charset=utf-8" },
  });
}

import anthropic
from dataclasses import dataclass, field
from datetime import datetime
import json
 
client = anthropic.Anthropic()
 
@dataclass
class RequestMetrics:
    timestamp: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int
    cache_write_tokens: int
    latency_ms: float
    model: str
    cost_usd: float
 
    def to_dict(self):
        return {
            "timestamp": self.timestamp,
            "input_tokens": self.input_tokens,
            "output_tokens": self.output_tokens,
            "cache_read_tokens": self.cache_read_tokens,
            "cache_write_tokens": self.cache_write_tokens,
            "latency_ms": self.latency_ms,
            "model": self.model,
            "cost_usd": self.cost_usd,
        }
 
def compute_cost(usage, model: str = "claude-sonnet-4-6") -> float:
    """Compute actual cost in USD based on token usage."""
    # Sonnet 4.6 pricing per 1M tokens
    INPUT_RATE = 3.00 / 1_000_000
    OUTPUT_RATE = 15.00 / 1_000_000
    CACHE_READ_RATE = 0.30 / 1_000_000
    CACHE_WRITE_RATE = 3.75 / 1_000_000
 
    cache_read = getattr(usage, "cache_read_input_tokens", 0)
    cache_write = getattr(usage, "cache_creation_input_tokens", 0)
    standard_input = usage.input_tokens - cache_read - cache_write
 
    return (
        standard_input * INPUT_RATE
        + usage.output_tokens * OUTPUT_RATE
        + cache_read * CACHE_READ_RATE
        + cache_write * CACHE_WRITE_RATE
    )
 
metrics_log: list[RequestMetrics] = []
 
def monitored_call(prompt: str, system: str = None) -> str:
    """Wrap any Sonnet 4.6 call with automatic metrics collection."""
    import time
 
    kwargs = {
        "model": "claude-sonnet-4-6",
        "max_tokens": 4096,
        "messages": [{"role": "user", "content": prompt}],
    }
 
    if system:
        kwargs["system"] = [
            {
                "type": "text",
                "text": system,
                "cache_control": {"type": "ephemeral"},
            }
        ]
 
    start = time.time()
    response = client.messages.create(**kwargs)
    latency = (time.time() - start) * 1000
 
    usage = response.usage
    metrics = RequestMetrics(
        timestamp=datetime.utcnow().isoformat(),
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cache_read_tokens=getattr(usage, "cache_read_input_tokens", 0),
        cache_write_tokens=getattr(usage, "cache_creation_input_tokens", 0),
        latency_ms=round(latency, 2),
        model="claude-sonnet-4-6",
        cost_usd=round(compute_cost(usage), 6),
    )
    metrics_log.append(metrics)
 
    # Log to your observability stack (Datadog, Grafana, CloudWatch, etc.)
    print(json.dumps(metrics.to_dict()))
 
    return first_text(response)
 
def print_usage_summary() -> None:
    """Print aggregated stats across all requests in this session."""
    if not metrics_log:
        return
    total_cost = sum(m.cost_usd for m in metrics_log)
    total_input = sum(m.input_tokens for m in metrics_log)
    total_cache_reads = sum(m.cache_read_tokens for m in metrics_log)
    avg_latency = sum(m.latency_ms for m in metrics_log) / len(metrics_log)
    cache_hit_rate = (
        total_cache_reads / total_input * 100 if total_input > 0 else 0
    )
 
    print(f"\n=== Session Summary ===")
    print(f"Requests:      {len(metrics_log)}")
    print(f"Total cost:    ${total_cost:.4f}")
    print(f"Avg latency:   {avg_latency:.0f}ms")
    print(f"Cache hit rate:{cache_hit_rate:.1f}%")
    print(f"Total tokens:  {total_input + sum(m.output_tokens for m in metrics_log):,}")

import hashlib
 
def debug_log(prompt: str, response_text: str, thinking: str = None) -> None:
    """Write a structured log entry for quality auditing."""
    entry = {
        "request_hash": hashlib.sha256(prompt.encode()).hexdigest()[:8],
        "prompt_length": len(prompt),
        "response_length": len(response_text),
        "thinking_length": len(thinking) if thinking else 0,
        "has_thinking": thinking is not None,
        # Truncate for log size management
        "prompt_preview": prompt[:200],
        "response_preview": response_text[:200],
    }
    print(json.dumps(entry))

# Before
response = client.messages.create(
    model="claude-sonnet-4-5",
    ...
)
 
# After — no other changes required in the vast majority of cases
response = client.messages.create(
    model="claude-sonnet-4-6",
    ...
)

import anthropic
 
client = anthropic.Anthropic()
 
PRICES = {  # $/MTok (input, output)
    "claude-sonnet-4-6": (3.0, 15.0),
    "claude-sonnet-5": (2.0, 10.0),
}
 
 
def compare_input_cost(system_prompt: str, user_message: str) -> None:
    """Count the same prompt on both models and print input-side cost."""
    for model, (in_price, _out_price) in PRICES.items():
        counted = client.messages.count_tokens(
            model=model,
            system=system_prompt,
            messages=[{"role": "user", "content": user_message}],
        )
        tokens = counted.input_tokens
        cost = tokens * in_price / 1_000_000
        print(f"{model:22} {tokens:>8,} tokens  ${cost:.6f}")
 
 
# Expected output (illustrative):
# claude-sonnet-4-6         12,480 tokens  $0.037440
# claude-sonnet-5           16,120 tokens  $0.032240

def smart_truncate(history: list, max_tokens: int = 900_000) -> list:
    """Remove oldest turns when approaching the context limit."""
    estimated = sum(len(m["content"]) // 4 for m in history)
    while estimated > max_tokens and len(history) > 2:
        history.pop(0)  # Remove oldest user turn
        history.pop(0)  # Remove oldest assistant turn
        estimated = sum(len(m["content"]) // 4 for m in history)
    return history

import time
import random
 
def resilient_call(prompt: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4096,
                messages=[{"role": "user", "content": prompt}],
            )
            return first_text(response)
        except anthropic.APIStatusError as e:
            if e.status_code == 529 and attempt < max_retries - 1:
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"API overloaded. Retrying in {wait:.1f}s ({attempt+1}/{max_retries})")
                time.sleep(wait)
            else:
                raise

# Wrong: max_tokens too low
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=5000,    # Not enough when budget_tokens=10000
    thinking={"type": "enabled", "budget_tokens": 10000},
    ...
)
 
# Correct: max_tokens = thinking budget + output headroom
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=18000,   # 10000 thinking + 8000 output
    thinking={"type": "enabled", "budget_tokens": 10000},
    ...
)