Claude Agent SDK em produção: CI/CD, servidores e app embarcado

# Python
pip install claude-agent-sdk

# TypeScript
npm install @anthropic-ai/claude-agent-sdk

# Python — query minimal
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    async for message in query(
        prompt="Liste os arquivos Python em src/ e sugira os que poderiam ser divididos",
        options=ClaudeAgentOptions(
            cwd="/path/to/project",
            allowed_tools=["Read", "Glob", "Grep"],
            model="claude-opus-4-7",
            effort="high",
        ),
    ):
        # message types: system, tool_use, tool_result, text, result
        if message.type == "text":
            print(message.text)
        elif message.type == "result":
            print(f"---\nDone. Total cost: $\{message.total_cost_usd}")

asyncio.run(main())

// TypeScript
import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Revise o código em src/auth.ts por vulnerabilidades OWASP",
  options: {
    cwd: "/path/to/project",
    allowedTools: ["Read", "Glob", "Grep"],
    model: "claude-opus-4-7",
    effort: "high",
  },
})) {
  if (message.type === "text") {
    console.log(message.text);
  } else if (message.type === "result") {
    console.log(`Total cost: $${message.total_cost_usd}`);
  }
}

# Python — custom tool via decorator
from claude_agent_sdk import query, tool, ClaudeAgentOptions
import httpx

@tool
async def fetch_jira_ticket(ticket_id: str) -> dict:
    """Busca detalhes de um ticket Jira.

    Args:
        ticket_id: ID do ticket no formato PROJ-123
    """
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"https://your-domain.atlassian.net/rest/api/3/issue/\{ticket_id}",
            headers={"Authorization": "Bearer $\{JIRA_TOKEN}"},
        )
        data = r.json()
        return {
            "summary": data["fields"]["summary"],
            "status": data["fields"]["status"]["name"],
            "description": data["fields"]["description"],
        }

@tool
async def post_to_slack(channel: str, message: str) -> str:
    """Envia mensagem para um canal Slack."""
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://slack.com/api/chat.postMessage",
            headers={"Authorization": "Bearer $\{SLACK_TOKEN}"},
            json={"channel": channel, "text": message},
        )
    return "Posted"

# Usar custom tools:
async for message in query(
    prompt="Verifique o ticket PROJ-123 e poste resumo no canal #eng",
    options=ClaudeAgentOptions(
        tools=[fetch_jira_ticket, post_to_slack],  # registra E disponibiliza
        allowed_tools=[                              # pre-aprova (sem permission prompt)
            "fetch_jira_ticket",
            "post_to_slack",
            "Read",
        ],
    ),
):
    ...

from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AgentDefinition,
    HookCallback,
    HookMatcher,
)

# Subagent definido programaticamente
code_reviewer = AgentDefinition(
    description="Reviewer de código focado em segurança",
    prompt="""Você é um revisor de código sênior.
Analise o código por: injection, auth bugs, IDOR, XSS.
Retorne APENAS problemas encontrados em formato estruturado.""",
    tools=["Read", "Glob", "Grep"],
    model="claude-opus-4-7",
    effort="high",
)

# Hook programático com callback Python
async def audit_edits(input_data, tool_use_id, context):
    """Loga toda edição de arquivo em audit.jsonl"""
    file_path = input_data.get("tool_input", {}).get("file_path", "?")
    from datetime import datetime
    import json
    entry = {
        "ts": datetime.utcnow().isoformat() + "Z",
        "tool_use_id": tool_use_id,
        "file_path": file_path,
        "session_id": context.get("session_id"),
    }
    with open("audit.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")
    return {}  # não bloqueia

async def block_env_edits(input_data, tool_use_id, context):
    """Bloqueia edição em .env*"""
    file_path = input_data.get("tool_input", {}).get("file_path", "")
    if ".env" in file_path:
        return {
            "decision": "block",
            "reason": "Edição em arquivos .env é proibida por policy",
        }
    return {}

async for message in query(
    prompt="Revise o PR #456 e aplique fixes de segurança sugeridos",
    options=ClaudeAgentOptions(
        agents={
            "code-reviewer": code_reviewer,  # subagent disponível
        },
        hooks={
            "PostToolUse": [
                HookMatcher(matcher="Edit|Write", hooks=[audit_edits])
            ],
            "PreToolUse": [
                HookMatcher(matcher="Edit", hooks=[block_env_edits])
            ],
        },
        allowed_tools=["Read", "Edit", "Agent"],
    ),
):
    ...

// TypeScript — conectar MCP server programaticamente
import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Navegue para github.com/anthropics/claude-code e resuma o README",
  options: {
    mcpServers: {
      playwright: {
        command: "npx",
        args: ["@playwright/mcp@latest"],
      },
      filesystem: {
        command: "npx",
        args: ["@modelcontextprotocol/server-filesystem", "/tmp"],
      },
    },
    allowedTools: [
      "Read",
      "mcp__playwright__navigate",
      "mcp__playwright__screenshot",
      "mcp__filesystem__write_file",
    ],
  },
})) {
  if (message.type === "text") console.log(message.text);
}

# Captura session_id na primeira chamada
session_id = None

async for message in query(prompt="Explore src/auth.ts", options=ClaudeAgentOptions(...)):
    if message.type == "system" and message.subtype == "init":
        session_id = message.data["session_id"]
        print(f"Session: \{session_id}")
    if message.type == "text":
        print(message.text)

# Salva em algum lugar persistente
with open(".claude/session-id", "w") as f:
    f.write(session_id)

# --- Processo 2 (mais tarde, talvez em outro step do CI) ---

with open(".claude/session-id") as f:
    session_id = f.read().strip()

# Resume: continua de onde parou
async for message in query(
    prompt="Agora liste todos os callers de authenticateUser",
    options=ClaudeAgentOptions(resume=session_id),
):
    ...

# Fork: novo session_id a partir do estado atual (não sobrescreve)
async for message in query(
    prompt="Experimenta abordagem diferente: refatore pra JWT",
    options=ClaudeAgentOptions(resume=session_id, fork_session=True),
):
    ...

# .github/workflows/ai-review.yml
name: Claude Code Review
on:
  pull_request:
    types: [opened, synchronize, reopened]

jobs:
  review:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install claude-agent-sdk httpx
      - name: Run Claude Review
        env:
          ANTHROPIC_API_KEY: ${'$'}{{ secrets.ANTHROPIC_API_KEY }}
          GITHUB_TOKEN: ${'$'}{{ secrets.GITHUB_TOKEN }}
          PR_NUMBER: ${'$'}{{ github.event.pull_request.number }}
          REPO: ${'$'}{{ github.repository }}
        run: python .github/scripts/review.py

# .github/scripts/review.py
import asyncio, os, httpx
from claude_agent_sdk import query, ClaudeAgentOptions

async def fetch_pr_diff(repo: str, pr_number: int, token: str) -> str:
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"https://api.github.com/repos/\{repo}/pulls/\{pr_number}",
            headers={"Accept": "application/vnd.github.v3.diff",
                     "Authorization": f"Bearer \{token}"},
        )
        return r.text

async def post_review(repo: str, pr: int, token: str, detail: str):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"https://api.github.com/repos/\{repo}/issues/\{pr}/comments",
            headers={"Authorization": f"Bearer \{token}",
                     "Accept": "application/vnd.github.v3+json"},
            json={"body": body},
        )

async def main():
    diff = await fetch_pr_diff(
        os.environ["REPO"],
        int(os.environ["PR_NUMBER"]),
        os.environ["GITHUB_TOKEN"],
    )

    review_body = []
    async for message in query(
        prompt=f"""Revise este PR focando em: segurança, performance, edge cases.
Retorne em markdown com seções: ✅ Positivos, ⚠️ Atenção, 🔴 Bloqueadores.
Seja objetivo e cite arquivo:linha quando possível.

---
\{diff}
""",
        options=ClaudeAgentOptions(
            model="claude-opus-4-7",
            effort="high",
            allowed_tools=["Read"],
            max_turns=5,
        ),
    ):
        if message.type == "text":
            review_body.append(message.text)

    body = "## 🤖 Claude Code Review\n\n" + "".join(review_body)
    await post_review(os.environ["REPO"], int(os.environ["PR_NUMBER"]),
                      os.environ["GITHUB_TOKEN"], body)

if __name__ == "__main__":
    asyncio.run(main())

// scripts/pr-triage.ts
// Roda como cron: agrega PRs abertos, classifica, atualiza board
import { query } from "@anthropic-ai/claude-agent-sdk";
import { Octokit } from "octokit";

const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });

interface Triage {
  pr_number: number;
  priority: "P0" | "P1" | "P2" | "P3";
  category: "bug" | "feature" | "chore" | "security";
  estimated_review_minutes: number;
  blockers: string[];
}

async function triagePR(owner: string, repo: string, pr: number): Promise<Triage> {
  const { data: prData } = await octokit.rest.pulls.get({ owner, repo, pull_number: pr });
  const diff = await octokit.rest.pulls.get({
    owner, repo, pull_number: pr,
    mediaType: { format: "diff" },
  });

  let result = "";
  for await (const message of query({
    prompt: `Classifique este PR em JSON strict (sem markdown fences):
{
  "priority": "P0|P1|P2|P3",
  "category": "bug|feature|chore|security",
  "estimated_review_minutes": <number>,
  "blockers": ["lista de bloqueadores detectados"]
}

PR: ${prData.title}
Descrição: ${prData.body ?? ""}
Diff:
${diff.data}
`,
    options: {
      model: "claude-haiku-4-5",  // rápido + barato para triage
      allowedTools: [],
      maxTurns: 1,
      jsonSchema: {
        type: "object",
        properties: {
          priority: { type: "string", enum: ["P0", "P1", "P2", "P3"] },
          category: { type: "string" },
          estimated_review_minutes: { type: "number" },
          blockers: { type: "array", items: { type: "string" } },
        },
        required: ["priority", "category", "estimated_review_minutes", "blockers"],
      },
    },
  })) {
    if (message.type === "text") result += message.text;
  }

  return { pr_number: pr, ...JSON.parse(result) };
}

async function main() {
  const owner = "empresa";
  const repo = "backend";
  const prs = await octokit.rest.pulls.list({ owner, repo, state: "open" });

  const triages = await Promise.all(
    prs.data.map(pr => triagePR(owner, repo, pr.number))
  );

  // Atualiza labels no GitHub baseado na triagem
  for (const t of triages) {
    await octokit.rest.issues.setLabels({
      owner, repo, issue_number: t.pr_number,
      labels: [`priority/${t.priority}`, `category/${t.category}`],
    });
  }

  // Notifica Slack com sumário
  const summary = triages
    .filter(t => t.priority === "P0" || t.priority === "P1")
    .map(t => `• PR #${t.pr_number} (${t.priority}): ${t.estimated_review_minutes}min`)
    .join("\n");

  if (summary) {
    await fetch(process.env.SLACK_WEBHOOK!, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ text: `🚨 PRs prioritários:\n${summary}` }),
    });
  }
}

main().catch(console.error);