Anthropic API, Claude Agent SDK, Tool Use, Computer Use & MCP

import os
import anthropic

# Boa prática (Cap. 17): centralize o identificador do modelo em um único ponto
DEFAULT_MODEL = os.getenv("CLAUDE_MODEL", "claude-sonnet-4-6")

client = anthropic.Anthropic()  # lê ANTHROPIC_API_KEY do ambiente

response = client.messages.create(
    model=DEFAULT_MODEL,
    max_tokens=600,
    temperature=0.2,
    system=(
        "Você é um engenheiro de suporte nível 3 especialista em PostgreSQL e ERP. "
        "Responda em português do Brasil com diagnóstico objetivo e SQL seguro."
    ),
    messages=[
        {
            "role": "user",
            "content": "Uma query em tabela de 8 milhões de linhas ficou lenta após filtro por JSONB. Como investigar e indexar?"
        }
    ]
)

print("Stop reason:", response.stop_reason)
print("Tokens (in/out):", response.usage.input_tokens, response.usage.output_tokens)
print(response.content[0].text)

import anthropic

client = anthropic.Anthropic()
MANUAL_SISTEMA = open("manual_regras_folha.txt", encoding="utf-8").read()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=800,
    system=[
        {
            "type": "text",
            "text": "Você analisa chamados com base estrita no manual técnico abaixo:\n\n" + MANUAL_SISTEMA,
            "cache_control": {"type": "ephemeral"}  # reutiliza o prefixo pesado no cache!
        }
    ],
    messages=[{"role": "user", "content": "Qual o código de evento para rescisão complementar?"}]
) as stream:
    for chunk in stream.text_stream:
        print(chunk, end="", flush=True)

import json
import anthropic

client = anthropic.Anthropic()

tools = [{
    "name": "get_exchange_rate",
    "description": (
        "Consulta a taxa de câmbio comercial atual entre duas moedas ISO (ex.: USD, BRL, EUR, CLP). "
        "Use sempre que o usuário pedir conversão de valores. Nunca adivinhe taxas de memória."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "from_currency": {"type": "string", "description": "Código ISO de 3 letras, ex.: 'USD'"},
            "to_currency": {"type": "string", "description": "Código ISO de 3 letras, ex.: 'BRL'"}
        },
        "required": ["from_currency", "to_currency"]
    }
}]

messages = [{"role": "user", "content": "Quanto custam 180 USD em BRL hoje?"}]
resp = client.messages.create(model="claude-sonnet-4-6", max_tokens=512, tools=tools, messages=messages)

if resp.stop_reason == "tool_use":
    messages.append({"role": "assistant", "content": resp.content})
    tool_results = []
    for block in resp.content:
        if block.type == "tool_use":
            # Seu código executa a integração real (ou devolve is_error=True explicativo se falhar)
            try:
                rate = 5.18  # Exemplo: retorno da sua API de câmbio
                payload = json.dumps({"pair": f"{block.input['from_currency']}/{block.input['to_currency']}", "rate": rate})
                tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": payload})
            except Exception as err:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "is_error": True,
                    "content": f"Falha temporária na API de câmbio ({err}). Informe o usuário sem inventar valores."
                })
    messages.append({"role": "user", "content": tool_results})
    final_resp = client.messages.create(model="claude-sonnet-4-6", max_tokens=512, tools=tools, messages=messages)
    print(final_resp.content[0].text)

import base64
import anthropic

client = anthropic.Anthropic()
img_b64 = base64.standard_b64encode(open("nota_fiscal.png", "rb").read()).decode()

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=600,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {"type": "base64", "media_type": "image/png", "data": img_b64}
            },
            {
                "type": "text",
                "text": (
                    "Extraia numero_nota, data_emissao, cnpj_emissor e valor_total. "
                    "Retorne exclusivamente um objeto JSON válido; se algum campo estiver ilegível, use null."
                )
            }
        ]
    }]
)
print(resp.content[0].text)

# Padrão arquitetural do Capítulo 18 (confirme a assinatura atual em docs.claude.com)
READ_ONLY_TOOLS = {"consultar_logs", "buscar_documentacao", "listar_chamados"}

def can_run(tool_name: str, args: dict) -> bool:
    """Ferramentas de leitura rodam livremente; escrita/exclusão exige aprovação humana."""
    if tool_name in READ_ONLY_TOOLS:
        return True
    resposta = input(f"[APROVAÇÃO HUMANA] Permitir {tool_name} com args={args}? (s/N): ")
    return resposta.strip().lower() == "s"

# Configure sempre max_iterations e observabilidade de eventos (on_event)
agent_config = {
    "model": "claude-sonnet-4-6",
    "max_iterations": 10,
    "permission_gate": can_run
}

// 1. Trecho do claude_desktop_config.json (Apêndice D)
{
  "mcpServers": {
    "postgres-leitura": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://leitor_ro:senha@localhost:5432/erp_db"]
    },
    "diagnostico-interno": {
      "command": "python",
      "args": ["/caminho/para/servidor_mcp.py"]
    }
  }
}

# 2. servidor_mcp.py — Servidor MCP customizado em Python (FastMCP)
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("diagnostico-interno")

@mcp.tool()
def verificar_fila_eventos(empresa_id: int) -> str:
    """Retorna a contagem de eventos pendentes na fila (somente leitura)."""
    return f"Empresa {empresa_id}: 0 erros críticos, 3 eventos aguardando processamento."

if __name__ == "__main__":
    mcp.run()