Claude Opus 4 Deep Dive: The Complete Anthropic 2026 Developer Ecosystem Guide

import anthropic

client = anthropic.Anthropic()

# Use Extended Thinking mode to solve complex architectural problems
response = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000  # Allocate token budget for internal reasoning
    },
    messages=[{
        "role": "user",
        "content": """Design a real-time messaging system architecture
        that supports tens of millions of concurrent connections.
        Consider: message persistence, ordering guarantees,
        cross-datacenter synchronization, and client reconnection strategies.
        Provide a detailed architecture diagram and technology choices."""
    }]
)

# Output includes both reasoning process and final answer
for block in response.content:
    if block.type == "thinking":
        print(f"[Reasoning] {block.thinking}")
    elif block.type == "text":
        print(f"[Answer] {block.text}")

import anthropic
import json

client = anthropic.Anthropic()

# Define a set of tools
tools = [
    {
        "name": "query_database",
        "description": "Query the product database with SQL syntax",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "SQL query statement"},
                "database": {"type": "string", "enum": ["products", "orders", "users"]}
            },
            "required": ["sql", "database"]
        }
    },
    {
        "name": "generate_chart",
        "description": "Generate a visualization chart from data",
        "input_schema": {
            "type": "object",
            "properties": {
                "chart_type": {"type": "string", "enum": ["bar", "line", "pie", "scatter"]},
                "data": {"type": "object", "description": "Chart data"},
                "title": {"type": "string"}
            },
            "required": ["chart_type", "data"]
        }
    },
    {
        "name": "send_report",
        "description": "Send a report via email or Slack",
        "input_schema": {
            "type": "object",
            "properties": {
                "channel": {"type": "string", "enum": ["email", "slack"]},
                "recipient": {"type": "string"},
                "content": {"type": "string"},
                "attachments": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["channel", "recipient", "content"]
        }
    }
]

# Agent loop: handle multi-step tasks
def run_agent(user_message: str):
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-20250509",
            max_tokens=8000,
            tools=tools,
            messages=messages
        )

        # Check if tools need to be executed
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })

            # Feed tool results back to the model
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            # Model has completed the task
            return response.content[0].text

def execute_tool(name: str, params: dict) -> dict:
    # In production, this would call real tools
    if name == "query_database":
        return {"rows": [{"product": "Widget A", "sales": 15000}, {"product": "Widget B", "sales": 12000}]}
    elif name == "generate_chart":
        return {"chart_url": "https://charts.example.com/abc123.png"}
    elif name == "send_report":
        return {"status": "sent", "message_id": "msg_789"}

# Run the Agent
result = run_agent("""Analyze the product sales data from the past 30 days,
generate a trend chart, and send the report to team@example.com""")
print(result)

from anthropic import Anthropic
import base64

client = Anthropic()

# Use Computer Use to automate tasks
response = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=10000,
    tools=[
        {
            "type": "computer_20250124",
            "name": "computer",
            "display_width_px": 1920,
            "display_height_px": 1080,
            "display_number": 0
        }
    ],
    messages=[{
        "role": "user",
        "content": """Open a browser, go to GitHub.com,
        search for the 'anthropic claude' repository,
        navigate to the latest official SDK repo and read the README."""
    }]
)

from anthropic.swarm import Swarm, Agent

# Define specialized Agents
researcher = Agent(
    name="Researcher",
    instructions="""You are a technical researcher. Your task is to search for
    and summarize the latest technical documentation. Use search tools to
    gather information and organize results in a structured format.""",
    tools=["web_search", "document_reader"],
    model="claude-opus-4-20250509"
)

coder = Agent(
    name="Coder",
    instructions="""You are a senior software engineer. Write high-quality code
    based on requirements documentation. Follow best practices, include
    complete error handling and type annotations.""",
    tools=["code_executor", "file_manager"],
    model="claude-sonnet-4-20250509"  # Use cost-effective Sonnet for coding tasks
)

reviewer = Agent(
    name="Reviewer",
    instructions="""You are a code review expert. Review code for security,
    performance, and maintainability. Provide specific improvement
    suggestions with example code.""",
    tools=["code_executor"],
    model="claude-opus-4-20250509"
)

# Create a Swarm and run the collaboration workflow
swarm = Swarm(agents=[researcher, coder, reviewer])

result = swarm.run(
    starting_agent=researcher,
    message="""Research a comparison of the most popular Python web frameworks,
    then write a high-performance REST API benchmarking tool,
    and finally conduct a code review and optimize.""",
    max_turns=15
)

print(result.final_output)

import anthropic

client = anthropic.Anthropic()

# Create a batch processing job
batch = client.batches.create(
    requests=[
        {
            "custom_id": f"doc-{i}",
            "params": {
                "model": "claude-opus-4-20250509",
                "max_tokens": 4000,
                "messages": [{"role": "user", "content": f"Analyze the following document and extract key information: {doc}"}]
            }
        }
        for i, doc in enumerate(documents)
    ]
)

# Check batch job status
status = client.batches.retrieve(batch.id)
print(f"Progress: {status.request_counts.completed}/{status.request_counts.total}")

# Retrieve results
results = client.batches.results(batch.id)
for result in results:
    print(f"{result.custom_id}: {result.result.message.content[0].text}")

import anthropic

client = anthropic.Anthropic()

# Optimize multi-turn conversations with Prompt Caching
system_prompt = """You are a professional legal advisor specializing in
Chinese civil and commercial law. Below are the legal provisions and
case references you need to consult:
[... extensive legal documentation ...]"""  # ~50K tokens

# First request establishes the cache
response1 = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=4000,
    system=[
        {
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}  # Mark as cacheable
        }
    ],
    messages=[{"role": "user", "content": "Explain the doctrine of changed circumstances in contract law"}]
)

# Subsequent requests hit the cache, significantly reducing costs
response2 = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=4000,
    system=[
        {
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[
        {"role": "user", "content": "Explain the doctrine of changed circumstances in contract law"},
        {"role": "assistant", "content": response1.content[0].text},
        {"role": "user", "content": "What is the difference between force majeure and changed circumstances?"}
    ]
)

import anthropic

client = anthropic.Anthropic()

def smart_agent(task: str) -> str:
    """Hybrid Agent using Opus for planning and Sonnet for execution"""

    # Step 1: Generate an execution plan with Opus
    plan_response = client.messages.create(
        model="claude-opus-4-20250509",
        max_tokens=4000,
        thinking={"type": "enabled", "budget_tokens": 5000},
        messages=[{
            "role": "user",
            "content": f"""Generate a detailed execution plan for the following task.
            Output a JSON-formatted list of steps, each containing:
            action, description, tools_needed.

            Task: {task}"""
        }]
    )

    plan = plan_response.content[-1].text  # Get the final text output

    # Step 2: Execute the plan step-by-step with Sonnet
    execution_response = client.messages.create(
        model="claude-sonnet-4-20250509",
        max_tokens=8000,
        tools=available_tools,
        messages=[{
            "role": "user",
            "content": f"""Execute the following task according to the plan:

            Plan:
            {plan}

            Original task: {task}

            Execute strictly according to the plan steps and report progress after each step."""
        }]
    )

    return execution_response.content[0].text

# Install the latest Claude Code
npm install -g @anthropic-ai/claude-code@latest

# Start Claude Code in a project
cd my-project
claude

# Example interaction:
# > Refactor all service classes in the src/services/ directory,
# > converting them from inheritance patterns to composition patterns,
# > and ensure all existing tests pass

# Claude Code will:
# 1. Analyze existing code structure
# 2. Generate a refactoring plan
# 3. Modify each file step by step
# 4. Run tests to verify
# 5. Automatically create a git commit

import anthropic
from anthropic.types import MessageStreamEvent

client = anthropic.Anthropic()

# Type-safe streaming output
with client.messages.stream(
    model="claude-opus-4-20250509",
    max_tokens=8000,
    messages=[{"role": "user", "content": "Explain the basic principles of quantum computing"}]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            print(f"\n[Done] Tokens used: {event.usage}")

# Structured output constraints
from pydantic import BaseModel

class CodeReview(BaseModel):
    summary: str
    issues: list[dict]
    score: int
    recommendations: list[str]

response = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=4000,
    messages=[{"role": "user", "content": f"Review the following code:\n{code}"}],
    response_format={
        "type": "json_schema",
        "schema": CodeReview.model_json_schema()
    }
)

review = CodeReview.model_validate_json(response.content[0].text)
print(f"Code score: {review.score}/10")

import anthropic
import json
from pathlib import Path
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class AnalysisResult:
    summary: str
    key_points: list[str]
    risks: list[dict]
    action_items: list[str]
    confidence: float

class DocumentAnalysisAgent:
    """Enterprise document analysis Agent with multi-document cross-analysis"""

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.model = "claude-opus-4-20250509"
        self.analysis_cache = {}

    def analyze_document(self, doc_path: str, context: str = "") -> AnalysisResult:
        """Analyze a single document"""
        content = Path(doc_path).read_text(encoding="utf-8")

        response = self.client.messages.create(
            model=self.model,
            max_tokens=8000,
            thinking={"type": "enabled", "budget_tokens": 6000},
            system="""You are a senior enterprise document analyst.
            When analyzing documents, focus on:
            1. Extracting core insights and key data points
            2. Identifying potential risks and compliance issues
            3. Generating actionable recommendations
            4. Assessing the confidence level of your analysis""",
            messages=[{
                "role": "user",
                "content": f"""Analyze the following document:

            {content}

            {f'Additional context: {context}' if context else ''}

            Output the analysis results in JSON format."""
            }],
            response_format={
                "type": "json_schema",
                "schema": {
                    "type": "object",
                    "properties": {
                        "summary": {"type": "string"},
                        "key_points": {"type": "array", "items": {"type": "string"}},
                        "risks": {"type": "array", "items": {
                            "type": "object",
                            "properties": {
                                "description": {"type": "string"},
                                "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]}
                            }
                        }},
                        "action_items": {"type": "array", "items": {"type": "string"}},
                        "confidence": {"type": "number", "minimum": 0, "maximum": 1}
                    }
                }
            }
        )

        result = json.loads(response.content[-1].text)
        return AnalysisResult(**result)

    def cross_analyze(self, results: list[AnalysisResult]) -> str:
        """Cross-reference analysis results from multiple documents"""
        combined = json.dumps([vars(r) for r in results], ensure_ascii=False, indent=2)

        response = self.client.messages.create(
            model=self.model,
            max_tokens=6000,
            messages=[{
                "role": "user",
                "content": f"""Below are independent analysis results from multiple documents.
            Please perform a cross-analysis to identify:
            1. Common themes and trends
            2. Contradictions or inconsistencies
            3. Comprehensive risk assessment
            4. Prioritized action recommendations

            Analysis results:
            {combined}"""
            }]
        )

        return response.content[0].text

# Usage example
agent = DocumentAnalysisAgent()

# Analyze multiple contract documents
contracts = ["contract_a.txt", "contract_b.txt", "contract_c.txt"]
results = [agent.analyze_document(c, context="M&A due diligence") for c in contracts]

# Cross-analyze
insights = agent.cross_analyze(results)
print(insights)

# Before (Claude 3.5 Sonnet)
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=4096,
    messages=[{"role": "user", "content": "..."}]
)

# After (Claude Opus 4)
response = client.messages.create(
    model="claude-opus-4-20250509",
    max_tokens=16000,  # Increase recommended; Opus 4 supports larger outputs
    thinking={"type": "enabled", "budget_tokens": 8000},  # Optional: enable Extended Thinking
    messages=[{"role": "user", "content": "..."}]
)