API da Anthropic: messages, streaming, vision, batch e cache

# SDK Python — pip install anthropic
import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

# Request básico
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system="Você é um assistente técnico especialista em Python. Responda em PT-BR.",
    messages=[
        {"role": "user", "content": "Explique o GIL em 3 frases"}
    ]
)

print(response.content[0].text)
print(f"Tokens usados: {response.usage.input_tokens} in, {response.usage.output_tokens} out")

# Conversa multi-turno
messages = []
messages.append({"role": "user", "content": "O que é o GIL?"})
# (chamada à API retorna resposta)
messages.append({"role": "assistant", "content": response.content[0].text})
messages.append({"role": "user", "content": "E como isso afeta o asyncio?"})
# Nova chamada com o histórico completo

# IMPORTANTE: a API é stateless — você é responsável por enviar
# o histórico completo de mensagens a cada request.
# Não há sessão no servidor. Todo o estado fica no cliente.

# Streaming: o modelo envia tokens à medida que gera
import anthropic

client = anthropic.Anthropic()

# Context manager para streaming
with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Escreva um poema sobre o GIL"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()  # nova linha ao final

# Com eventos completos (para saber quando termina, ver stop_reason, etc.)
with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_stop":
            print(f"

Stop reason: {stream.get_final_message().stop_reason}")

# SDK TypeScript/JavaScript:
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();

const stream = await client.messages.stream({
  model: "claude-sonnet-4-5",
  max_tokens: 1024,
  messages: [{ role: "user", content: "Hello" }],
});

for await (const chunk of stream) {
  if (chunk.type === "content_block_delta") {
    process.stdout.write(chunk.delta.text);
  }
}

# Para Next.js API Routes (streaming SSE):
export async function POST(req: Request) {
  const encoder = new TextEncoder();
  const stream = new TransformStream();
  const writer = stream.writable.getWriter();

  const anthropicStream = await client.messages.stream({ ... });

  (async () => {
    for await (const chunk of anthropicStream) {
      if (chunk.type === "content_block_delta") {
        await writer.write(encoder.encode(`data: ${chunk.delta.text}

`));
      }
    }
    await writer.close();
  })();

  return new Response(stream.readable, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

# Claude Sonnet e Opus suportam imagens (Haiku tem suporte limitado)
# Formatos: JPEG, PNG, GIF, WebP
# Limite: até 20 imagens por request; cada imagem ≤ 5MB

import anthropic, base64

client = anthropic.Anthropic()

# Opção 1: URL pública (mais simples)
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "url",
                    "url": "https://exemplo.com/screenshot.png"
                }
            },
            {
                "type": "text",
                "text": "Descreva o que está nesta imagem e identifique eventuais problemas de UI."
            }
        ]
    }]
)

# Opção 2: Base64 (para imagens locais ou sem URL pública)
with open("screenshot.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": image_data
                }
            },
            {"type": "text", "text": "Existe algum erro ou warning neste screenshot de log?"}
        ]
    }]
)
print(response.content[0].text)

# Custo de imagem: aproximadamente 1.600 tokens base + tokens por dimensão
# 1080x1080 px ≈ 1.600 tokens ≈ $0.005 com Sonnet (input)
# Otimização: redimensione imagens para max 1.568px no lado maior antes de enviar

# Prompt caching: marque seções que não mudam entre requests

# Caso ideal: system prompt longo + documentos de referência + pergunta do usuário
# O system + documentos ficam cacheados; apenas a pergunta varia

import anthropic

client = anthropic.Anthropic()

# Documento de referência longo (1000+ tokens) que é o mesmo em todos os requests
MANUAL_TECNICO = """
# Manual Técnico do Sistema de Pagamentos
[... milhares de linhas de documentação ...]
"""

def perguntar_sobre_manual(pergunta: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "Você é um assistente especialista no manual técnico abaixo.",
                "cache_control": {"type": "ephemeral"}  # ← cacheia esta parte
            },
            {
                "type": "text",
                "text": MANUAL_TECNICO,
                "cache_control": {"type": "ephemeral"}  # ← cacheia o manual também
            }
        ],
        messages=[{"role": "user", "content": pergunta}]  # ← não cacheado (varia)
    )

    # Verificar se o cache foi usado:
    usage = response.usage
    print(f"Input normal: {usage.input_tokens} tokens")
    print(f"Input cacheado: {usage.cache_read_input_tokens} tokens (10% do preço)")
    print(f"Cache criado: {usage.cache_creation_input_tokens} tokens (110% do preço, uma vez)")

    return response.content[0].text

# Primeiro request: paga 110% para criar o cache (uma vez)
# Requests subsequentes (dentro de 5min): paga 10% para a parte cacheada
# Economia real em 100 requests: ~80-90% do custo de input

# Messages Batch API — para alto volume sem urgência de latência

import anthropic, json
from datetime import datetime

client = anthropic.Anthropic()

# Criar um batch de requests
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"analise-{i}",   # ID único para rastrear o resultado
            "params": {
                "model": "claude-haiku-4-5",  # Haiku + batch = custo mínimo
                "max_tokens": 500,
                "messages": [{
                    "role": "user",
                    "content": f"Classifique o sentimento deste review: {review}"
                }]
            }
        }
        for i, review in enumerate(lista_de_reviews)  # até 10.000 por batch
    ]
)

print(f"Batch criado: {batch.id}")
print(f"Status: {batch.processing_status}")  # "in_progress"

# Verificar status (polling — geralmente termina em minutos)
import time
while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    print(f"Aguardando... {batch.request_counts}")
    time.sleep(30)

# Processar resultados
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        texto = result.result.message.content[0].text
        print(f"{result.custom_id}: {texto}")
    else:
        print(f"{result.custom_id}: ERRO - {result.result.error}")