Arquitetura de
Agentes e Claude
Code na Prática

import anthropic, json
from typing import List

client = anthropic.Anthropic()

class PlanExecuteAgent:
    """
    Fase 1: Opus 4.6 cria plano detalhado
    Fase 2: Sonnet 4.6 executa cada etapa
    Resultado: mais barato e mais auditável que ReAct puro
    """

    def plan(self, objective: str, available_tools: list) -> List[dict]:
        # Planner usa modelo mais capaz para raciocínio
        tools_desc = json.dumps([t["name"] for t in available_tools])
        response = client.messages.create(
            model="claude-opus-4-6",   # Modelo mais poderoso para planejamento
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Crie um plano de execução detalhado para:
                
OBJETIVO: {objective}

FERRAMENTAS DISPONÍVEIS: {tools_desc}

Responda APENAS com JSON válido:
{{
  "plano": [
    {{
      "etapa": 1,
      "descricao": "descrição clara da etapa",
      "tool": "nome_da_tool_ou_null",
      "input_esperado": "o que esta etapa precisa de entrada",
      "output_esperado": "o que esta etapa deve produzir",
      "pode_falhar": true/false,
      "acao_se_falhar": "retry/skip/abort"
    }}
  ]
}}"""
            }]
        )
        plan_data = json.loads(response.content[0].text)
        return plan_data["plano"]

    def execute(self, plan: List[dict], tools: list) -> dict:
        results = {}
        execution_log = []

        for step in plan:
            print(f"▶ Etapa {step['etapa']}: {step['descricao']}")

            # Construir contexto com resultados anteriores
            context = json.dumps({
                k: v for k, v in results.items()
            })

            try:
                # Executor usa modelo mais rápido/barato
                response = client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=1024,
                    tools=tools,
                    messages=[{
                        "role": "user",
                        "content": f"""Execute esta etapa do plano:

ETAPA: {step['descricao']}
RESULTADO ESPERADO: {step['output_esperado']}
CONTEXTO (etapas anteriores): {context}

Execute a etapa usando as ferramentas disponíveis."""
                    }]
                )
                # Processar tool calls se necessário
                result = self._process_response(response, tools)
                results[f"etapa_{step['etapa']}"] = result
                execution_log.append({"etapa": step["etapa"], "status": "ok", "result": result})

            except Exception as e:
                if step.get("acao_se_falhar") == "abort":
                    raise RuntimeError(f"Etapa {step['etapa']} falhou (abort): {e}")
                execution_log.append({"etapa": step["etapa"], "status": "failed", "error": str(e)})

        return {"results": results, "log": execution_log}

    def _process_response(self, response, tools):
        # Processar tool calls (mesmo padrão do M7)
        if response.stop_reason == "end_turn":
            return response.content[0].text
        # ... loop de tool use (ver M7)
        return "[tool result processado]"

import anthropic, asyncio, json
from dataclasses import dataclass
from typing import Dict, Any

client = anthropic.Anthropic()

@dataclass
class SubAgent:
    name: str
    model: str
    system_prompt: str
    tools: list

# Definição dos sub-agentes especializados
AGENTS = {
    "fiscal": SubAgent(
        name="Agente Fiscal",
        model="claude-sonnet-4-6",
        system_prompt="Especialista em análise fiscal, NF-e e obrigações tributárias.",
        tools=FISCAL_TOOLS  # definidos no projeto
    ),
    "rh": SubAgent(
        name="Agente RH",
        model="claude-sonnet-4-6",
        system_prompt="Especialista em folha de pagamento, CLT e gestão de pessoas.",
        tools=RH_TOOLS
    ),
    "docs": SubAgent(
        name="Agente Documentação",
        model="claude-haiku-4-5",  # Haiku para tarefas simples = economia
        system_prompt="Recupera e sumariza documentos corporativos.",
        tools=DOC_TOOLS
    ),
}

async def run_subagent(agent_key: str, task: str) -> Dict[str, Any]:
    """Executa um sub-agente de forma assíncrona."""
    agent = AGENTS[agent_key]

    # Em produção: usar asyncio com httpx ou anthropic async client
    response = client.messages.create(
        model=agent.model,
        max_tokens=2048,
        system=agent.system_prompt,
        tools=agent.tools,
        messages=[{"role": "user", "content": task}]
    )
    return {
        "agent": agent.name,
        "result": response.content[0].text,
        "tokens_used": response.usage.input_tokens + response.usage.output_tokens
    }

class SupervisorAgent:
    def run(self, objective: str) -> str:
        # 1. Supervisor decompõe o objetivo em tarefas por agente
        decomp = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""Decomponha este objetivo em tarefas para agentes especializados.
                
OBJETIVO: {objective}
AGENTES DISPONÍVEIS: fiscal (NF-e, SPED), rh (folha, CLT), docs (documentos)

Responda APENAS com JSON:
{{"tarefas": [{{"agente": "fiscal|rh|docs", "tarefa": "descrição precisa"}}]}}"""
            }]
        )
        tarefas = json.loads(decomp.content[0].text)["tarefas"]

        # 2. Executar sub-agentes em paralelo
        loop = asyncio.new_event_loop()
        tasks = [run_subagent(t["agente"], t["tarefa"]) for t in tarefas]
        results = loop.run_until_complete(asyncio.gather(*tasks))

        # 3. Supervisor consolida todos os resultados
        results_text = json.dumps(results, ensure_ascii=False, indent=2)
        final = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            messages=[{
                "role": "user",
                "content": f"""Consolide os resultados dos agentes em relatório executivo.

OBJETIVO ORIGINAL: {objective}
RESULTADOS DOS AGENTES:
{results_text}

Produza um relatório claro, destacando achados importantes e recomendações."""
            }]
        )
        return final.content[0].text

# CLAUDE.md — FP-MAIN v2 — Configuração Avançada

## Contexto do Projeto
Sistema de folha IBM z/OS 2.5, COBOL IBM Enterprise 6.3, DB2 v13.
Python 3.11 (FastAPI) como camada de integração.

## Comandos Customizados (slash commands)
- /analisa-cobol [arquivo] → Analisa programa COBOL, lista divisões e paragraphs
- /gera-teste [programa]  → Gera casos de teste para programa COBOL
- /doc-cobol [arquivo]    → Gera documentação Markdown do programa
- /diff-seguro            → Mostra diff e pede confirmação antes de salvar

## Hooks de Segurança
ANTES de qualquer edição em *.cbl ou *.cpy:
  1. Verificar se arquivo tem backup recente (git status)
  2. Alertar se arquivo é chamado por mais de 5 programas (alto impacto)
  3. NUNCA editar arquivos em PROD — verificar se estamos no branch correto

## Regras de Edição COBOL
- Manter colunas 1-6 em branco (área de sequência)
- Indicador na coluna 7 (* para comentário)
- Código nas colunas 8-72 (área A e B)
- Colunas 73-80: identificação (reservadas, não tocar)
- Todo novo PARAGRAPH deve ter comentário de propósito

## Sub-agentes Disponíveis
Para tarefas grandes, spawnar sub-agentes:
- Análise: "analise todos os programas com GOTO e liste refatorações"
- Testes: "gere testes unitários para os 20 programas sem cobertura"
- Docs: "documente todos os copybooks do diretório /copybooks"

## Workflow de Sessão Longa
1. Ao iniciar: ler docs/decisoes/ADRs-recentes.md
2. A cada 10 ações: resumir progresso em .claude/session-progress.md
3. Ao finalizar: atualizar docs/changelog-ai.md com o que foi feito

name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with: {fetch-depth: 0}

      - name: Install Claude Code
        run: npm install -g @anthropic-ai/claude-code

      - name: AI Review do diff
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          # Gerar diff do PR
          git diff origin/main...HEAD > /tmp/pr.diff

          # Claude Code revisa e gera comentários
          claude -p "Revise este diff de código COBOL/Python. \
            Identifique: (1) bugs potenciais, (2) violações das \
            convenções do CLAUDE.md, (3) edge cases não tratados. \
            Formate como lista de comentários com linha e severidade." \
            --input /tmp/pr.diff \
            --output /tmp/review.md

      - name: Post review como comentário no PR
        uses: actions/github-script@v7
        with:
          script: |
            const review = require('fs').readFileSync('/tmp/review.md', 'utf8');
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## 🤖 AI Code Review\n\n${review}`
            });

import anthropic, base64
from PIL import ImageGrab
import subprocess

client = anthropic.Anthropic()

def screenshot_to_base64() -> str:
    img = ImageGrab.grab()
    import io
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode()

def computer_use_agent(task: str, max_steps: int = 20):
    """
    Agente que opera terminal 3270 (ISPF/TSO) via Computer Use.
    Caso de uso real: extrair dados de datasets COBOL sem API disponível.
    """
    messages = []
    steps = 0

    while steps < max_steps:
        # Capturar estado atual da tela
        screen = screenshot_to_base64()

        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/png", "data": screen}
                },
                {"type": "text", "content": task if steps == 0 else "Continue a tarefa."}
            ]
        })

        response = client.beta.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            tools=[{"type": "computer_20250124", "name": "computer",
                    "display_width_px": 1920, "display_height_px": 1080}],
            betas=["computer-use-2025-01-24"],
            messages=messages
        )

        for block in response.content:
            if block.type == "tool_use" and block.name == "computer":
                action = block.input["action"]

                # HITL: aprovar ações que escrevem dados
                if action in ["type", "key"] and "ENTER" in str(block.input):
                    confirm = input(f"⚠️ Claude vai pressionar ENTER. Confirmar? (s/n): ")
                    if confirm.lower() != "s":
                        return "Interrompido pelo usuário."

                # Executar ação via pyautogui ou xdotool
                self._execute_computer_action(block.input)

            elif block.type == "text" and "TAREFA CONCLUÍDA" in block.text:
                return block.text

        steps += 1

    return "Limite de steps atingido"

from plan_execute_agent import PlanExecuteAgent
from memory.manager import MemoryManager
from tools.mainframe import MAINFRAME_TOOLS
import anthropic, json

client = anthropic.Anthropic()
memory = MemoryManager()

SYSTEM_FOLHA = """Você é o Agente de Fechamento de Folha da Empresa XYZ.

RESPONSABILIDADES:
- Executar o fechamento mensal seguindo o plano aprovado
- Identificar e escalar exceções para revisão humana
- NUNCA aprovar pagamento sem HITL explícito
- Manter log detalhado de cada ação para auditoria

REGRAS DE SEGURANÇA:
- Ações irreversíveis (pagamento, emissão de DARF) requerem aprovação humana
- Divergências > R$ 1.000 são exceções obrigatórias
- Qualquer SQLCODE de erro: escalar imediatamente"""

class FolhaAgent:
    def run_fechamento(self, competencia: str) -> dict:
        # Recuperar contexto de meses anteriores (memória episódica)
        historico = memory.get_episode_summary("folha_fechamento", last_n=3)

        # Fase 1: Planejamento com contexto histórico
        planner = PlanExecuteAgent()
        objetivo = f"""Fechar folha de pagamento competência {competencia}.
        
Histórico: {historico}

Etapas obrigatórias:
1. Verificar pré-requisitos (pontos lançados, admissões/demissões do mês)
2. Calcular folha via FPFOLHA00 para todos os departamentos
3. Validar totais vs. mês anterior (alertar divergências > 5%)
4. Processar encargos (INSS, FGTS, IRRF) via programas específicos
5. Identificar exceções e agrupar por tipo
6. Gerar relatórios gerenciais
7. AGUARDAR APROVAÇÃO HUMANA antes de qualquer pagamento"""

        plano = planner.plan(objetivo, MAINFRAME_TOOLS)

        # Mostrar plano para aprovação humana (HITL fase 1)
        print("\n📋 PLANO DE FECHAMENTO:")
        for etapa in plano:
            print(f"  {etapa['etapa']}. {etapa['descricao']}")

        confirm = input("\n✅ Aprovar plano? (s/n): ")
        if confirm.lower() != "s":
            return {"status": "cancelado"}

        # Fase 2: Execução — parar em etapas que precisam de HITL
        resultados = planner.execute(plano, MAINFRAME_TOOLS)

        # Fase 3: Análise de exceções
        excecoes = self._analisar_excecoes(resultados)

        if excecoes:
            print(f"\n⚠️ {len(excecoes)} exceções encontradas:")
            for exc in excecoes:
                print(f"  • {exc['descricao']} — {exc['valor']}")
            confirm2 = input("\n✅ Exceções revisadas. Aprovar fechamento? (s/n): ")
            if confirm2.lower() != "s":
                return {"status": "pendente_revisao", "excecoes": excecoes}

        # Fase 4: Pagamento (requer HITL final)
        print("\n🏦 AÇÃO IRREVERSÍVEL: Autorizar pagamento via TED/PIX")
        confirm3 = input("CONFIRMAR PAGAMENTO? Digite 'CONFIRMO': ")
        if confirm3 != "CONFIRMO":
            return {"status": "pagamento_cancelado"}

        # Fase 5: Pagamento + atualizar memória episódica
        resultado_final = self._executar_pagamento(competencia)
        memory.add_episode("folha_fechamento", {
            "competencia": competencia,
            "excecoes": len(excecoes),
            "status": "concluido"
        })

        return resultado_final

import time, json, uuid
from dataclasses import dataclass, field
from typing import List, Any
from contextlib import contextmanager
import anthropic

@dataclass
class AgentTrace:
    session_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    objective: str = ""
    steps: List[dict] = field(default_factory=list)
    started_at: float = field(default_factory=time.time)
    total_tokens: int = 0
    total_cost_usd: float = 0.0

    def add_step(self, step_type: str, data: Any, duration_ms: float = 0):
        self.steps.append({
            "type": step_type,
            "data": data,
            "timestamp": time.time() - self.started_at,
            "duration_ms": duration_ms
        })

    def to_json(self) -> str:
        return json.dumps({
            "session_id": self.session_id,
            "objective": self.objective,
            "duration_s": round(time.time() - self.started_at, 2),
            "steps_count": len(self.steps),
            "total_tokens": self.total_tokens,
            "total_cost_usd": round(self.total_cost_usd, 4),
            "steps": self.steps
        }, ensure_ascii=False, indent=2)

class TracedAgent:
    """Wrapper que adiciona observabilidade a qualquer agente."""

    COST_PER_MTOK = {  # USD por milhão de tokens (abr/2026)
        "claude-opus-4-6":    {"input": 15.0,  "output": 75.0},
        "claude-sonnet-4-6":  {"input":  3.0,  "output": 15.0},
        "claude-haiku-4-5":   {"input":  0.8,  "output":  4.0},
    }

    def __init__(self, agent_fn, objective: str):
        self.agent_fn  = agent_fn
        self.trace     = AgentTrace(objective=objective)
        self.client    = anthropic.Anthropic()

    def traced_api_call(self, **kwargs):
        t0 = time.time()
        response = self.client.messages.create(**kwargs)
        duration = (time.time() - t0) * 1000

        # Registrar no trace
        model = kwargs.get("model", "unknown")
        costs = self.COST_PER_MTOK.get(model, {"input": 0, "output": 0})
        cost = (response.usage.input_tokens  * costs["input"]  / 1_000_000 +
                response.usage.output_tokens * costs["output"] / 1_000_000)

        self.trace.total_tokens += response.usage.input_tokens + response.usage.output_tokens
        self.trace.total_cost_usd += cost

        self.trace.add_step("api_call", {
            "model": model,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cost_usd": round(cost, 5),
            "stop_reason": response.stop_reason
        }, duration)

        # Alerta de custo anômalo
        if self.trace.total_cost_usd > 1.0:
            print(f"⚠️ CUSTO ALTO: USD {self.trace.total_cost_usd:.3f} na sessão {self.trace.session_id}")

        return response

    def save_trace(self, path: str = None):
        path = path or f"traces/{self.trace.session_id}.json"
        with open(path, "w") as f:
            f.write(self.trace.to_json())
        print(f"📊 Trace salvo: {path}")