Capítulo 3: La Función query() — Fundamentos
Capítulo 3: La Función query() — Fundamentos
1. ¿Qué es query()?
query() es el punto de entrada principal del SDK. Toda interacción con Claude Code pasa por esta función. Cuando la invocas, el SDK hace lo siguiente en secuencia:
- Serializa tus opciones y el prompt en una llamada al proceso
claudeCLI - Spawna el proceso Claude Code como subprocess
- Establece una comunicación bidireccional via stdin/stdout con formato JSON
- Lee el stream de mensajes que Claude Code emite mientras trabaja
- Deserializa cada mensaje y lo convierte en un objeto tipado
- Yield de cada mensaje al caller mediante un async generator/iterable
La función no bloquea. Emite mensajes a medida que Claude trabaja, permitiendo que tu aplicación reaccione en tiempo real: mostrar progreso, loguear herramientas usadas, actualizar una UI, o simplemente esperar el resultado final.
Por qué es un async generator
Claude Code es un proceso agentic que puede ejecutar múltiples pasos antes de terminar. Una query que pide “refactoriza todos los archivos TypeScript del proyecto” puede durar varios minutos y emitir decenas de mensajes intermedios. Usar un async generator es la solución natural porque:
- Backpressure: el caller controla cuándo pedir el siguiente mensaje
- Memory efficiency: no acumula todos los mensajes en memoria
- Early exit: puedes hacer
breaksi detectas lo que necesitabas - Streaming UX: puedes mostrar texto del asistente antes de que termine
Diagrama: ciclo de vida de query()
sequenceDiagram
participant App as Tu Aplicación
participant SDK as SDK (query)
participant CLI as Claude Code CLI
participant API as Anthropic API
App->>SDK: query(prompt, options)
SDK->>CLI: spawn proceso claude
CLI->>API: POST /messages
API-->>CLI: stream de tokens
CLI-->>SDK: SystemMessage (init)
SDK-->>App: yield SystemMessage
CLI-->>SDK: AssistantMessage (texto/tool_use)
SDK-->>App: yield AssistantMessage
CLI->>CLI: ejecuta herramienta (Read/Bash/etc)
CLI-->>SDK: ToolResultMessage
SDK-->>App: yield ToolResultMessage
CLI-->>SDK: AssistantMessage (análisis)
SDK-->>App: yield AssistantMessage
CLI-->>SDK: ResultMessage (final)
SDK-->>App: yield ResultMessage
SDK->>CLI: proceso terminaSignature completa en Python
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import Message
from typing import AsyncGenerator
async def query(
prompt: str,
options: ClaudeCodeOptions | None = None,
) -> AsyncGenerator[Message, None]:
...El return type es AsyncGenerator[Message, None] donde Message es un union type de los cuatro tipos posibles. En la práctica, iteras con async for.
Signature completa en TypeScript
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
import type { Message } from "@anthropic-ai/claude-code-sdk";
declare function query(options: {
prompt: string;
options?: ClaudeCodeOptions;
}): AsyncIterable<Message>;En TypeScript la función recibe un objeto con prompt y options, mientras que Python recibe argumentos posicionales/keyword.
Ejemplo mínimo funcional
# Python
import asyncio
from claude_code_sdk import query
async def main():
async for message in query("¿Cuánto es 2 + 2?"):
print(message)
asyncio.run(main())// TypeScript
import { query } from "@anthropic-ai/claude-code-sdk";
async function main() {
for await (const message of query({ prompt: "¿Cuánto es 2 + 2?" })) {
console.log(message);
}
}
main();2. Tipos de mensajes del stream
El stream emite exactamente cuatro tipos de mensajes. Comprender cada uno es fundamental para construir aplicaciones robustas.
2.1 SystemMessage — Inicialización
El primer mensaje que siempre recibes. Contiene metadatos de la sesión.
Estructura:
# Python - tipo SystemMessage
{
"type": "system",
"subtype": "init",
"session_id": "sess_01XYZ...",
"tools": [
"Read", "Write", "Edit", "Bash",
"Glob", "Grep", "WebSearch", "WebFetch",
"Agent", "Task", "AskUserQuestion"
],
"mcp_servers": [],
"model": "claude-sonnet-4-5",
"api_key_source": "environment"
}// TypeScript - interface SystemMessage
interface SystemMessage {
type: "system";
subtype: "init";
session_id: string;
tools: string[];
mcp_servers: MCPServerInfo[];
model: string;
api_key_source: "environment" | "cli_flag" | "options";
}Uso típico: guardar el session_id para poder retomar la sesión después, verificar qué herramientas están disponibles, loguear el modelo que se está usando.
# Python - procesar SystemMessage
from claude_code_sdk.types import SystemMessage
async for message in query(prompt, options):
if isinstance(message, SystemMessage):
session_id = message.session_id
print(f"Sesión iniciada: {session_id}")
print(f"Herramientas disponibles: {message.tools}")
print(f"Modelo: {message.model}")// TypeScript - procesar SystemMessage
for await (const message of query({ prompt, options })) {
if (message.type === "system" && message.subtype === "init") {
const sessionId = message.session_id;
console.log(`Sesión iniciada: ${sessionId}`);
console.log(`Herramientas: ${message.tools.join(", ")}`);
}
}2.2 AssistantMessage — Respuesta del asistente
El mensaje más frecuente. Contiene el texto que Claude genera y/o bloques tool_use cuando Claude decide usar una herramienta.
Estructura con texto puro:
{
"type": "assistant",
"message": {
"id": "msg_01ABC...",
"type": "message",
"role": "assistant",
"content": [
{
"type": "text",
"text": "Voy a analizar los archivos del proyecto..."
}
],
"model": "claude-sonnet-4-5",
"stop_reason": "end_turn",
"usage": {
"input_tokens": 1250,
"output_tokens": 87
}
}
}Estructura con tool_use:
{
"type": "assistant",
"message": {
"id": "msg_02DEF...",
"role": "assistant",
"content": [
{
"type": "text",
"text": "Primero leeré el archivo principal."
},
{
"type": "tool_use",
"id": "toolu_01GHI...",
"name": "Read",
"input": {
"file_path": "/home/user/project/main.py"
}
}
],
"stop_reason": "tool_use"
}
}Procesamiento completo:
# Python
from claude_code_sdk.types import AssistantMessage
async for message in query(prompt, options):
if isinstance(message, AssistantMessage):
for block in message.message.content:
if block.type == "text":
# Texto del asistente — mostrar en tiempo real
print(block.text, end="", flush=True)
elif block.type == "tool_use":
# El agente va a usar una herramienta
print(f"\n[HERRAMIENTA] {block.name}")
print(f" Input: {block.input}")// TypeScript
for await (const message of query({ prompt, options })) {
if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "text") {
process.stdout.write(block.text);
} else if (block.type === "tool_use") {
console.log(`\n[HERRAMIENTA] ${block.name}`);
console.log(` Input:`, block.input);
}
}
}
}Interface TypeScript completa:
interface AssistantMessage {
type: "assistant";
message: {
id: string;
type: "message";
role: "assistant";
content: ContentBlock[];
model: string;
stop_reason: "end_turn" | "tool_use" | "max_tokens" | "stop_sequence";
stop_sequence: string | null;
usage: {
input_tokens: number;
output_tokens: number;
cache_creation_input_tokens?: number;
cache_read_input_tokens?: number;
};
};
}
type ContentBlock = TextBlock | ToolUseBlock;
interface TextBlock {
type: "text";
text: string;
}
interface ToolUseBlock {
type: "tool_use";
id: string;
name: string;
input: Record<string, unknown>;
}2.3 ToolResultMessage — Resultado de herramienta
Emitido después de que Claude Code ejecuta una herramienta. Contiene el resultado (éxito o error).
Estructura:
{
"type": "tool_result",
"tool_use_id": "toolu_01GHI...",
"tool_name": "Read",
"content": [
{
"type": "text",
"text": " 1→import asyncio\n 2→from pathlib import Path\n..."
}
],
"is_error": false
}Cuando hay error:
{
"type": "tool_result",
"tool_use_id": "toolu_02JKL...",
"tool_name": "Bash",
"content": [
{
"type": "text",
"text": "Error: command not found: pytest"
}
],
"is_error": true
}Procesamiento:
# Python
from claude_code_sdk.types import ToolResultMessage
async for message in query(prompt, options):
if isinstance(message, ToolResultMessage):
status = "ERROR" if message.is_error else "OK"
print(f"[{status}] {message.tool_name}")
if message.is_error:
for block in message.content:
print(f" Error: {block.text}")// TypeScript
interface ToolResultMessage {
type: "tool_result";
tool_use_id: string;
tool_name: string;
content: Array<{ type: "text"; text: string }>;
is_error: boolean;
}2.4 ResultMessage — Mensaje final
El último mensaje del stream. Contiene el resultado final de la query, estadísticas de costo, duración, y el session_id para continuar la conversación.
Estructura completa:
{
"type": "result",
"subtype": "success",
"result": "He analizado el proyecto. Encontré 3 archivos con potenciales bugs:\n1. src/auth.py línea 45...",
"session_id": "sess_01XYZ...",
"cost_usd": 0.00234,
"duration_ms": 12450,
"duration_api_ms": 8930,
"num_turns": 4,
"is_error": false,
"total_cost_usd": 0.00234,
"usage": {
"input_tokens": 4521,
"output_tokens": 389,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 2100
}
}Cuando falla:
{
"type": "result",
"subtype": "error_max_turns",
"result": "Se alcanzó el límite de turns (10) sin completar la tarea.",
"session_id": "sess_01XYZ...",
"is_error": true,
"num_turns": 10
}Los posibles subtype son:
success: completó exitosamenteerror_max_turns: alcanzó el límite demax_turnserror_during_execution: error interno durante la ejecuciónerror_api: error de la API de Anthropic
Procesamiento:
# Python
from claude_code_sdk.types import ResultMessage
async for message in query(prompt, options):
if isinstance(message, ResultMessage):
if message.is_error:
print(f"Error ({message.subtype}): {message.result}")
else:
print(f"Resultado: {message.result}")
print(f"Costo: ${message.cost_usd:.4f}")
print(f"Duración: {message.duration_ms}ms")
print(f"Turns usados: {message.num_turns}")// TypeScript
interface ResultMessage {
type: "result";
subtype: "success" | "error_max_turns" | "error_during_execution" | "error_api";
result: string;
session_id: string;
cost_usd: number;
duration_ms: number;
duration_api_ms: number;
num_turns: number;
is_error: boolean;
total_cost_usd: number;
usage: {
input_tokens: number;
output_tokens: number;
cache_creation_input_tokens: number;
cache_read_input_tokens: number;
};
}Diagrama del flujo completo de mensajes
flowchart TD
A[query llamada] --> B[SystemMessage: init]
B --> C{¿Necesita herramienta?}
C -->|No| D[AssistantMessage: texto]
C -->|Sí| E[AssistantMessage: tool_use]
E --> F[ToolResultMessage: resultado]
F --> G{¿Tarea completa?}
G -->|No| C
G -->|Sí| H[AssistantMessage: respuesta final]
D --> I[ResultMessage: éxito/error]
H --> I3. ClaudeCodeOptions — Referencia completa
ClaudeCodeOptions controla todos los aspectos del comportamiento del agente. Conocer cada opción es esencial para construir agentes robustos y seguros.
Tabla completa de opciones
| Opción | Tipo | Default | Descripción |
|---|---|---|---|
allowed_tools | string[] | todas | Herramientas habilitadas |
system_prompt | string | ninguno | Instrucciones del sistema |
cwd | string | process.cwd() | Directorio de trabajo |
model | string | sonnet | Modelo de Claude |
max_turns | number | 10 | Máximo de iteraciones |
permission_mode | string | default | Modo de permisos |
resume | string | ninguno | Session ID para continuar |
mcp_servers | object[] | [] | Servidores MCP |
hooks | object | {} | Callbacks de eventos |
env | object | {} | Variables de entorno extra |
api_key | string | ANTHROPIC_API_KEY | Override de API key |
3.1 allowed_tools
Controla exactamente qué herramientas puede usar el agente. Principio de menor privilegio: solo habilita lo que realmente necesitas.
# Python
from claude_code_sdk import query, ClaudeCodeOptions
# Solo lectura — seguro para análisis
options = ClaudeCodeOptions(
allowed_tools=["Read", "Glob", "Grep"]
)
# Desarrollo activo — permite modificar archivos
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"]
)
# Investigación web
options = ClaudeCodeOptions(
allowed_tools=["WebSearch", "WebFetch", "Read"]
)
# Con herramienta MCP personalizada
options = ClaudeCodeOptions(
allowed_tools=["Read", "Grep", "mcp__database__query"]
)// TypeScript
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
// Solo lectura
const readOnlyOptions: ClaudeCodeOptions = {
allowed_tools: ["Read", "Glob", "Grep"],
};
// Desarrollo completo
const devOptions: ClaudeCodeOptions = {
allowed_tools: ["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
};Anti-patrón: dar todas las herramientas sin pensar
# MAL — el agente puede hacer cualquier cosa incluyendo rm -rf
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep",
"WebSearch", "WebFetch", "Agent", "Task"]
)
# BIEN — solo lo necesario para la tarea específica
options = ClaudeCodeOptions(
allowed_tools=["Read", "Grep", "Glob"] # tarea de análisis
)3.2 system_prompt
Instrucciones persistentes que el agente sigue en toda la sesión. No reemplaza el prompt del usuario, lo complementa.
# Python
system = """Eres un experto en seguridad de código Python.
Tu tarea es analizar código en busca de vulnerabilidades OWASP Top 10.
Para cada vulnerabilidad encontrada:
1. Indica el tipo (SQL Injection, XSS, etc.)
2. Señala la línea exacta
3. Explica el riesgo
4. Proporciona el código corregido
Formato de respuesta: JSON estructurado.
Idioma: español."""
options = ClaudeCodeOptions(
system_prompt=system,
allowed_tools=["Read", "Grep", "Glob"]
)// TypeScript
const securityOptions: ClaudeCodeOptions = {
system_prompt: `Eres un experto en seguridad de código Python.
Analiza SOLO vulnerabilidades, no hagas refactoring.
Responde siempre en JSON con el schema: {vulnerabilities: [{type, line, risk, fix}]}`,
allowed_tools: ["Read", "Grep", "Glob"],
};3.3 cwd (current working directory)
El directorio desde el que el agente “ve” el sistema de archivos. Las rutas relativas en las herramientas se resuelven desde aquí.
# Python
import os
from claude_code_sdk import query, ClaudeCodeOptions
# Analizar un proyecto específico
options = ClaudeCodeOptions(
cwd="/home/user/projects/mi-proyecto",
allowed_tools=["Read", "Glob", "Grep"]
)
async for message in query("Lista todos los archivos TypeScript", options):
...
# Usar el directorio actual del script
options = ClaudeCodeOptions(
cwd=os.getcwd(),
allowed_tools=["Read", "Glob"]
)
# Directorio temporal para operaciones de escritura seguras
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
options = ClaudeCodeOptions(
cwd=tmpdir,
allowed_tools=["Read", "Write", "Edit"],
permission_mode="bypassPermissions"
)// TypeScript
import path from "path";
const options: ClaudeCodeOptions = {
cwd: path.resolve("/home/user/projects/mi-proyecto"),
allowed_tools: ["Read", "Glob", "Grep"],
};Riesgo de cwd incorrecto: si el agente tiene herramienta Bash y el cwd apunta a /, puede explorar o modificar todo el sistema. Siempre usa el directorio más específico posible.
3.4 model
Selecciona el modelo de Claude. Cada modelo tiene diferentes capacidades y costos.
# Python
# Modelo potente para tareas complejas
options = ClaudeCodeOptions(
model="claude-opus-4-5",
max_turns=20
)
# Modelo balanceado (recomendado para la mayoría de casos)
options = ClaudeCodeOptions(
model="claude-sonnet-4-5"
)
# Modelo ligero para tareas simples
options = ClaudeCodeOptions(
model="claude-haiku-3-5",
max_turns=5
)// TypeScript
const opusOptions: ClaudeCodeOptions = {
model: "claude-opus-4-5",
max_turns: 25,
};
const sonnetOptions: ClaudeCodeOptions = {
model: "claude-sonnet-4-5", // default recomendado
};
const haikuOptions: ClaudeCodeOptions = {
model: "claude-haiku-3-5",
max_turns: 3,
};3.5 max_turns
Limita cuántas iteraciones del loop agentic puede hacer Claude. Un “turn” es un ciclo completo: AssistantMessage → herramienta → resultado → siguiente decisión.
# Python
# Tarea simple: no más de 3 pasos
options = ClaudeCodeOptions(
max_turns=3,
allowed_tools=["Read"]
)
# Refactoring complejo: muchos pasos necesarios
options = ClaudeCodeOptions(
max_turns=30,
allowed_tools=["Read", "Edit", "Bash"]
)
# Verificar cuántos turns se usaron
async for message in query(prompt, options):
if hasattr(message, 'num_turns'):
print(f"Turns usados: {message.num_turns}/{options.max_turns}")// TypeScript
const options: ClaudeCodeOptions = {
max_turns: 15,
};
for await (const message of query({ prompt, options })) {
if (message.type === "result") {
const pct = ((message.num_turns / 15) * 100).toFixed(0);
console.log(`Turns: ${message.num_turns}/15 (${pct}%)`);
}
}3.6 permission_mode
Controla cómo el agente maneja los permisos para ejecutar herramientas.
# Python
from claude_code_sdk import query, ClaudeCodeOptions
# default: Claude pide confirmación antes de editar/ejecutar
options = ClaudeCodeOptions(
permission_mode="default"
)
# acceptEdits: auto-acepta ediciones de archivos, pide OK para Bash
options = ClaudeCodeOptions(
permission_mode="acceptEdits"
)
# bypassPermissions: acepta todo sin preguntar (peligroso en producción)
options = ClaudeCodeOptions(
permission_mode="bypassPermissions"
)
# plan: solo planifica, no ejecuta nada
options = ClaudeCodeOptions(
permission_mode="plan"
)// TypeScript
type PermissionMode = "default" | "acceptEdits" | "bypassPermissions" | "plan";
const automatedOptions: ClaudeCodeOptions = {
permission_mode: "bypassPermissions", // para CI/CD automatizado
};
const safeOptions: ClaudeCodeOptions = {
permission_mode: "plan", // solo muestra el plan, no ejecuta
};3.7 resume
Continúa una sesión anterior usando su session_id. Permite conversaciones multi-turno donde el agente recuerda el contexto.
# Python
# Primera sesión
session_id = None
async for message in query("Analiza el archivo main.py"):
if hasattr(message, 'session_id') and message.type == "result":
session_id = message.session_id
# Continuar en una segunda query
if session_id:
options = ClaudeCodeOptions(resume=session_id)
async for message in query("Ahora refactoriza lo que analizaste", options):
...// TypeScript
let sessionId: string | undefined;
for await (const message of query({ prompt: "Analiza main.py" })) {
if (message.type === "result") {
sessionId = message.session_id;
}
}
// Continuar
if (sessionId) {
for await (const message of query({
prompt: "Ahora arregla los bugs que encontraste",
options: { resume: sessionId },
})) {
// ...
}
}3.8 mcp_servers
Configura servidores MCP (Model Context Protocol) adicionales que extienden las herramientas disponibles. Ver capítulo 5 para detalles completos.
# Python
options = ClaudeCodeOptions(
mcp_servers=[
{
"type": "stdio",
"command": "node",
"args": ["/path/to/my-mcp-server.js"],
"env": {"DATABASE_URL": "postgres://..."}
},
{
"type": "sse",
"url": "http://localhost:8080/mcp"
}
],
allowed_tools=["Read", "mcp__myserver__query_db"]
)// TypeScript
const options: ClaudeCodeOptions = {
mcp_servers: [
{
type: "stdio",
command: "node",
args: ["/path/to/mcp-server.js"],
},
],
allowed_tools: ["Read", "mcp__myserver__search"],
};3.9 env
Variables de entorno adicionales para el proceso Claude Code. Útil para pasar credenciales sin exponerlas en el prompt.
# Python
import os
options = ClaudeCodeOptions(
env={
"DATABASE_URL": os.environ["DATABASE_URL"],
"REDIS_URL": os.environ["REDIS_URL"],
"APP_ENV": "staging"
},
allowed_tools=["Read", "Bash"]
)// TypeScript
const options: ClaudeCodeOptions = {
env: {
DATABASE_URL: process.env.DATABASE_URL!,
NODE_ENV: "test",
},
};3.10 api_key
Override de la API key para casos donde no usas la variable de entorno ANTHROPIC_API_KEY.
# Python — útil para multi-tenant donde cada usuario tiene su key
options = ClaudeCodeOptions(
api_key=user.anthropic_api_key
)// TypeScript
const options: ClaudeCodeOptions = {
api_key: user.anthropicApiKey,
};Advertencia de seguridad: nunca loguees ni expongas la api_key. No la incluyas en mensajes de error ni en trazas de stack.
4. Procesamiento del stream — Patrones
4.1 Patrón básico: solo el resultado final
El caso más común: ejecutar una query y obtener solo la respuesta final.
# Python
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import ResultMessage
async def run_query(prompt: str, options: ClaudeCodeOptions | None = None) -> str:
"""Ejecuta una query y retorna solo el resultado final."""
async for message in query(prompt, options):
if isinstance(message, ResultMessage):
if message.is_error:
raise RuntimeError(f"Query falló ({message.subtype}): {message.result}")
return message.result
raise RuntimeError("Stream terminó sin ResultMessage")
# Uso
async def main():
result = await run_query(
"Cuenta cuántos archivos .py hay en el proyecto",
ClaudeCodeOptions(
cwd="/home/user/project",
allowed_tools=["Glob"]
)
)
print(result)// TypeScript
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
async function runQuery(
prompt: string,
options?: ClaudeCodeOptions
): Promise<string> {
for await (const message of query({ prompt, options })) {
if (message.type === "result") {
if (message.is_error) {
throw new Error(`Query falló (${message.subtype}): ${message.result}`);
}
return message.result;
}
}
throw new Error("Stream terminó sin ResultMessage");
}
// Uso
const result = await runQuery("Cuenta archivos .py", {
cwd: "/home/user/project",
allowed_tools: ["Glob"],
});
console.log(result);4.2 Patrón streaming: texto en tiempo real
Para interfaces de usuario que deben mostrar el texto a medida que se genera.
# Python
import asyncio
import sys
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import AssistantMessage, ResultMessage
async def stream_to_stdout(prompt: str, options: ClaudeCodeOptions | None = None):
"""Muestra el texto del asistente en tiempo real."""
async for message in query(prompt, options):
if isinstance(message, AssistantMessage):
for block in message.message.content:
if block.type == "text":
sys.stdout.write(block.text)
sys.stdout.flush()
elif isinstance(message, ResultMessage):
print(f"\n\n---\nCosto: ${message.cost_usd:.4f} | Turns: {message.num_turns}")
asyncio.run(stream_to_stdout("Explica qué hace este proyecto"))// TypeScript
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
async function streamToStdout(
prompt: string,
options?: ClaudeCodeOptions
): Promise<void> {
for await (const message of query({ prompt, options })) {
if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "text") {
process.stdout.write(block.text);
}
}
} else if (message.type === "result") {
console.log(
`\n\n---\nCosto: $${message.cost_usd.toFixed(4)} | Turns: ${message.num_turns}`
);
}
}
}4.3 Patrón auditoría: registrar herramientas usadas
Para compliance, debugging, y optimización de costos.
# Python
from dataclasses import dataclass, field
from datetime import datetime
from claude_code_sdk.types import AssistantMessage, ToolResultMessage, ResultMessage
@dataclass
class ToolCall:
name: str
input: dict
timestamp: datetime
success: bool = True
error: str | None = None
@dataclass
class QueryAudit:
session_id: str = ""
tool_calls: list[ToolCall] = field(default_factory=list)
cost_usd: float = 0.0
duration_ms: int = 0
num_turns: int = 0
result: str = ""
async def audited_query(prompt: str, options=None) -> QueryAudit:
audit = QueryAudit()
pending_tools: dict[str, ToolCall] = {}
async for message in query(prompt, options):
if message.type == "system":
audit.session_id = message.session_id
elif isinstance(message, AssistantMessage):
for block in message.message.content:
if block.type == "tool_use":
tc = ToolCall(
name=block.name,
input=block.input,
timestamp=datetime.now()
)
pending_tools[block.id] = tc
audit.tool_calls.append(tc)
elif isinstance(message, ToolResultMessage):
if message.tool_use_id in pending_tools:
tc = pending_tools[message.tool_use_id]
tc.success = not message.is_error
if message.is_error:
tc.error = message.content[0].text if message.content else "Error desconocido"
elif isinstance(message, ResultMessage):
audit.cost_usd = message.cost_usd
audit.duration_ms = message.duration_ms
audit.num_turns = message.num_turns
audit.result = message.result
return audit
# Uso
audit = await audited_query("Analiza el proyecto", options)
print(f"Herramientas usadas: {len(audit.tool_calls)}")
for tc in audit.tool_calls:
status = "OK" if tc.success else f"ERROR: {tc.error}"
print(f" {tc.name}: {status}")
print(f"Costo total: ${audit.cost_usd:.4f}")// TypeScript
interface ToolCall {
name: string;
input: Record<string, unknown>;
timestamp: Date;
success: boolean;
error?: string;
}
interface QueryAudit {
sessionId: string;
toolCalls: ToolCall[];
costUsd: number;
durationMs: number;
numTurns: number;
result: string;
}
async function auditedQuery(
prompt: string,
options?: ClaudeCodeOptions
): Promise<QueryAudit> {
const audit: QueryAudit = {
sessionId: "",
toolCalls: [],
costUsd: 0,
durationMs: 0,
numTurns: 0,
result: "",
};
const pendingTools = new Map<string, ToolCall>();
for await (const message of query({ prompt, options })) {
if (message.type === "system") {
audit.sessionId = message.session_id;
} else if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "tool_use") {
const tc: ToolCall = {
name: block.name,
input: block.input,
timestamp: new Date(),
success: true,
};
pendingTools.set(block.id, tc);
audit.toolCalls.push(tc);
}
}
} else if (message.type === "tool_result") {
const tc = pendingTools.get(message.tool_use_id);
if (tc) {
tc.success = !message.is_error;
if (message.is_error) {
tc.error = message.content[0]?.text ?? "Error desconocido";
}
}
} else if (message.type === "result") {
audit.costUsd = message.cost_usd;
audit.durationMs = message.duration_ms;
audit.numTurns = message.num_turns;
audit.result = message.result;
}
}
return audit;
}4.4 Patrón costo: monitoreo financiero
Para proyectos donde el costo de la API es crítico.
# Python
from claude_code_sdk.types import AssistantMessage, ResultMessage
async def cost_tracked_query(
prompt: str,
options=None,
budget_usd: float = 0.10
) -> tuple[str, float]:
"""Ejecuta query con límite de presupuesto."""
accumulated_cost = 0.0
input_tokens = 0
output_tokens = 0
async for message in query(prompt, options):
if isinstance(message, AssistantMessage):
usage = message.message.usage
input_tokens += usage.input_tokens
output_tokens += usage.output_tokens
# Estimación en tiempo real (precio Sonnet aproximado)
estimated_cost = (input_tokens * 3 + output_tokens * 15) / 1_000_000
if estimated_cost > budget_usd:
print(f"⚠ Presupuesto excedido: ${estimated_cost:.4f} > ${budget_usd}")
break
elif isinstance(message, ResultMessage):
accumulated_cost = message.cost_usd
return message.result, accumulated_cost
return "", accumulated_cost// TypeScript
async function costTrackedQuery(
prompt: string,
options?: ClaudeCodeOptions,
budgetUsd: number = 0.1
): Promise<{ result: string; costUsd: number }> {
let inputTokens = 0;
let outputTokens = 0;
for await (const message of query({ prompt, options })) {
if (message.type === "assistant") {
inputTokens += message.message.usage.input_tokens;
outputTokens += message.message.usage.output_tokens;
const estimatedCost = (inputTokens * 3 + outputTokens * 15) / 1_000_000;
if (estimatedCost > budgetUsd) {
console.warn(`Presupuesto excedido: $${estimatedCost.toFixed(4)}`);
break;
}
} else if (message.type === "result") {
return { result: message.result, costUsd: message.cost_usd };
}
}
return { result: "", costUsd: 0 };
}5. Manejo de errores
Tipos de errores
flowchart TD
E[Error en query] --> A{¿Tipo?}
A -->|CLI no instalada| B[CLINotFoundError]
A -->|Fallo de conexión| C[CLIConnectionError]
A -->|Proceso terminó| D[ProcessError]
A -->|Error API| F[APIError]
A -->|Timeout| G[TimeoutError]
B --> B1[Instalar claude CLI]
C --> C1[Verificar PATH, reintentar]
D --> D1[Revisar exit_code]
F --> F1[Verificar API key, rate limit]
G --> G1[Aumentar timeout o max_turns]CLINotFoundError
Ocurre cuando el binario claude no está en el PATH.
# Python
from claude_code_sdk import query
from claude_code_sdk.exceptions import CLINotFoundError
import subprocess
async def safe_query(prompt: str):
try:
async for message in query(prompt):
yield message
except CLINotFoundError:
# Intentar instalar
print("Claude CLI no encontrado. Instalando...")
subprocess.run(["npm", "install", "-g", "@anthropic-ai/claude-code"], check=True)
# Reintentar
async for message in query(prompt):
yield message// TypeScript
import { query } from "@anthropic-ai/claude-code-sdk";
import { CLINotFoundError } from "@anthropic-ai/claude-code-sdk";
import { execSync } from "child_process";
async function safeQuery(prompt: string) {
try {
for await (const message of query({ prompt })) {
yield message;
}
} catch (error) {
if (error instanceof CLINotFoundError) {
console.log("Instalando Claude CLI...");
execSync("npm install -g @anthropic-ai/claude-code");
for await (const message of query({ prompt })) {
yield message;
}
} else {
throw error;
}
}
}ProcessError
El proceso Claude Code terminó con un código de error.
# Python
from claude_code_sdk.exceptions import ProcessError
async def robust_query(prompt: str, max_retries: int = 3):
for attempt in range(max_retries):
try:
async for message in query(prompt):
yield message
return
except ProcessError as e:
print(f"Error de proceso (intento {attempt+1}): exit_code={e.exit_code}")
print(f"Stderr: {e.stderr}")
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt) # backoff exponencial// TypeScript
import { ProcessError } from "@anthropic-ai/claude-code-sdk";
async function robustQuery(
prompt: string,
maxRetries: number = 3
): Promise<string> {
for (let attempt = 0; attempt < maxRetries; attempt++) {
try {
for await (const message of query({ prompt })) {
if (message.type === "result") return message.result;
}
} catch (error) {
if (error instanceof ProcessError) {
console.error(`Intento ${attempt + 1}: exit_code=${error.exit_code}`);
if (attempt === maxRetries - 1) throw error;
await new Promise((r) => setTimeout(r, 1000 * 2 ** attempt));
} else {
throw error;
}
}
}
return "";
}Retry con backoff exponencial completo
# Python — implementación production-ready
import asyncio
import logging
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.exceptions import (
CLINotFoundError, CLIConnectionError,
ProcessError, APIError
)
logger = logging.getLogger(__name__)
async def production_query(
prompt: str,
options: ClaudeCodeOptions | None = None,
max_retries: int = 3,
base_delay: float = 1.0
) -> str:
"""Query con reintentos, backoff y logging estructurado."""
last_error = None
for attempt in range(max_retries):
try:
async for message in query(prompt, options):
if hasattr(message, 'is_error') and message.type == "result":
if message.is_error:
logger.error(
"query_error",
extra={
"subtype": message.subtype,
"result": message.result,
"session_id": message.session_id
}
)
raise RuntimeError(f"Query error: {message.result}")
return message.result
except CLINotFoundError:
logger.critical("claude_cli_not_found")
raise # No retry — necesita instalación manual
except CLIConnectionError as e:
last_error = e
delay = base_delay * (2 ** attempt)
logger.warning(
"cli_connection_error",
extra={"attempt": attempt + 1, "retry_in": delay}
)
await asyncio.sleep(delay)
except APIError as e:
last_error = e
if e.status_code == 429: # Rate limit
delay = base_delay * (2 ** attempt) * 2
logger.warning("rate_limited", extra={"retry_in": delay})
await asyncio.sleep(delay)
elif e.status_code >= 500: # Error de servidor
delay = base_delay * (2 ** attempt)
logger.warning("api_server_error", extra={"status": e.status_code})
await asyncio.sleep(delay)
else:
logger.error("api_client_error", extra={"status": e.status_code})
raise # 4xx no son retryables
except ProcessError as e:
last_error = e
logger.error("process_error", extra={"exit_code": e.exit_code})
if attempt < max_retries - 1:
await asyncio.sleep(base_delay)
raise RuntimeError(f"Query falló después de {max_retries} intentos") from last_error6. Control de contexto y cwd
El cwd (current working directory) es el “punto de vista” del agente. Desde él, todas las rutas relativas se resuelven.
Cómo el agente ve el filesystem
flowchart LR
subgraph "Sistema de archivos real"
ROOT["/"]
HOME["/home/user/"]
PROJ["/home/user/proyecto/"]
SRC["/home/user/proyecto/src/"]
F1["/home/user/proyecto/src/main.py"]
F2["/home/user/proyecto/tests/test_main.py"]
end
subgraph "Perspectiva del agente (cwd=/home/user/proyecto)"
A1["src/main.py"]
A2["tests/test_main.py"]
A3["./ (raíz del proyecto)"]
end
F1 -.->|"agente ve como"| A1
F2 -.->|"agente ve como"| A2
PROJ -.->|"agente ve como"| A3Ejemplos con diferentes cwd
# Python
# Análisis de proyecto específico
options = ClaudeCodeOptions(
cwd="/home/user/project-alpha",
allowed_tools=["Read", "Glob", "Grep"]
)
result = await run_query("¿Cuántos tests hay?", options)
# Múltiples proyectos en paralelo
import asyncio
async def analyze_project(path: str) -> dict:
opts = ClaudeCodeOptions(
cwd=path,
allowed_tools=["Read", "Glob", "Grep"],
model="claude-haiku-3-5"
)
result = await run_query("Lista las dependencias principales", opts)
return {"path": path, "deps": result}
projects = ["/proj/alpha", "/proj/beta", "/proj/gamma"]
results = await asyncio.gather(*[analyze_project(p) for p in projects])// TypeScript
import path from "path";
import os from "os";
import fs from "fs/promises";
// Directorio temporal seguro para operaciones de escritura
async function withTempDir<T>(
fn: (dir: string) => Promise<T>
): Promise<T> {
const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "claude-"));
try {
return await fn(tmpDir);
} finally {
await fs.rm(tmpDir, { recursive: true, force: true });
}
}
// Uso
const output = await withTempDir(async (tmpDir) => {
for await (const message of query({
prompt: "Genera un archivo README.md de ejemplo",
options: {
cwd: tmpDir,
allowed_tools: ["Write"],
permission_mode: "bypassPermissions",
},
})) {
if (message.type === "result") return message.result;
}
return "";
});Sandboxing del directorio
Para mayor seguridad en entornos de producción, considera usar un directorio sandbox que limite qué puede leer el agente.
# Python — copiar solo los archivos necesarios al sandbox
import shutil
import tempfile
from pathlib import Path
async def sandboxed_query(
source_dir: str,
prompt: str,
allowed_patterns: list[str] = ["*.py", "*.ts", "*.json"]
) -> str:
with tempfile.TemporaryDirectory() as sandbox:
# Copiar solo archivos relevantes
for pattern in allowed_patterns:
for file in Path(source_dir).rglob(pattern):
dest = Path(sandbox) / file.relative_to(source_dir)
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(file, dest)
opts = ClaudeCodeOptions(
cwd=sandbox,
allowed_tools=["Read", "Glob", "Grep"]
)
return await run_query(prompt, opts)7. system_prompt — Ingeniería de prompts
Diferencia entre system_prompt y prompt
| Aspecto | system_prompt | prompt (user message) |
|---|---|---|
| Persistencia | Toda la sesión | Solo el mensaje actual |
| Propósito | Definir comportamiento | Pedir tarea específica |
| Tokens | Siempre en contexto | Puede ser cacheado |
| Ejemplos | Few-shot permanente | Datos de la tarea |
Estructura recomendada
Un system prompt efectivo tiene cuatro secciones:
1. ROL: Quién eres
2. RESTRICCIONES: Qué no debes hacer
3. FORMATO: Cómo responder
4. FEW-SHOT: Ejemplos de comportamiento esperadoEjemplos por caso de uso
# Python — Agente de code review
code_review_system = """Eres un senior engineer especializado en Python con 10 años de experiencia.
Tu misión es hacer code review enfocado en:
- Correctness: bugs lógicos, edge cases
- Security: OWASP Top 10 para Python
- Performance: O(n) cuando podría ser O(1), queries N+1
- Maintainability: código duplicado, funciones largas
RESTRICCIONES:
- No sugieras refactorings cosméticos (nombres de variables, formato)
- No comentes sobre el estilo si está configurado en linters
- Solo Python >= 3.10
FORMATO DE RESPUESTA:
Siempre en JSON con este schema exacto:
{
"issues": [
{
"severity": "critical|high|medium|low",
"file": "path/to/file.py",
"line": 42,
"category": "security|performance|correctness|maintainability",
"description": "Descripción del problema",
"fix": "Código sugerido"
}
],
"summary": "Resumen general",
"score": 0-10
}"""
options = ClaudeCodeOptions(
system_prompt=code_review_system,
allowed_tools=["Read", "Grep", "Glob"],
model="claude-sonnet-4-5"
)# Python — Agente de documentación
doc_system = """Eres un technical writer experto en documentación de APIs.
Generas documentación clara, con ejemplos y en formato Markdown.
ESTILO:
- Usa headers H2 para secciones principales, H3 para subsecciones
- Incluye ejemplos de código en Python Y TypeScript para cada endpoint
- Escribe en inglés técnico, sin jerga
FEW-SHOT:
Input: función Python sin docstring que parsea JSON
Output:
## parse_json(data: str) -> dict
Parses a JSON string and returns a Python dictionary.
**Parameters:**
- `data` (str): Valid JSON string to parse.
**Returns:** `dict` — Parsed JSON as a Python dictionary.
**Raises:** `ValueError` if `data` is not valid JSON.
```python
result = parse_json('{"key": "value"}')
# {"key": "value"}"""
options = ClaudeCodeOptions( system_prompt=doc_system, allowed_tools=[“Read”, “Glob”, “Write”], permission_mode=“acceptEdits” )
```typescript
// TypeScript — Agente de testing
const testingSystem = `Eres un experto en testing de software con pytest y Jest.
Tu tarea es escribir tests unitarios completos.
PRINCIPIOS:
- AAA pattern: Arrange, Act, Assert
- Un assert por test cuando es posible
- Nombres descriptivos: test_should_<behavior>_when_<condition>
- Cobertura de: happy path, edge cases, error cases
RESTRICCIONES:
- No uses mocks a menos que sea absolutamente necesario
- Prefiere tests de integración sobre mocks excesivos
- Los tests deben ser deterministas (no depender de tiempo real o random)`;
const options: ClaudeCodeOptions = {
system_prompt: testingSystem,
allowed_tools: ["Read", "Write", "Glob", "Bash"],
permission_mode: "acceptEdits",
};Límites de tokens y estrategia de caché
El system prompt ocupa tokens en cada turn. Para prompts largos, Anthropic ofrece “prompt caching” que reduce el costo. El SDK lo gestiona automáticamente cuando el system prompt supera ~2000 tokens.
Anti-patrón: system prompt demasiado largo
# MAL — 5000 tokens que se repiten en cada turn
system = """
[toda la documentación de la API]
[todos los ejemplos de código]
[todas las reglas de estilo]
... 200 líneas ...
"""
# BIEN — system prompt conciso, datos en el prompt del usuario
system = """Eres un experto en la API interna. Las reglas están en CONTRIBUTING.md."""
# El agente lee CONTRIBUTING.md una vez con Read
options = ClaudeCodeOptions(
system_prompt=system,
allowed_tools=["Read"]
)
result = await run_query(
"Lee CONTRIBUTING.md y luego analiza src/auth.py",
options
)8. Modelos y selección
Tabla comparativa
| Modelo | Capacidad | Velocidad | Costo (input/output por 1M tokens) | Mejor para |
|---|---|---|---|---|
| claude-opus-4-5 | Máxima | Lento | ~$15 / $75 | Tareas muy complejas, razonamiento profundo |
| claude-sonnet-4-5 | Alta | Media | ~$3 / $15 | La mayoría de casos (recomendado) |
| claude-haiku-3-5 | Media | Muy rápido | ~$0.80 / $4 | Tareas simples, alto volumen |
Estrategia de selección dinámica
# Python — selección de modelo según complejidad estimada
from enum import Enum
class TaskComplexity(Enum):
SIMPLE = "simple" # grep, count, lookup
MEDIUM = "medium" # análisis, refactoring pequeño
COMPLEX = "complex" # arquitectura, migración grande
def select_model(complexity: TaskComplexity) -> str:
return {
TaskComplexity.SIMPLE: "claude-haiku-3-5",
TaskComplexity.MEDIUM: "claude-sonnet-4-5",
TaskComplexity.COMPLEX: "claude-opus-4-5",
}[complexity]
async def smart_query(prompt: str, complexity: TaskComplexity) -> str:
options = ClaudeCodeOptions(
model=select_model(complexity),
max_turns={
TaskComplexity.SIMPLE: 3,
TaskComplexity.MEDIUM: 10,
TaskComplexity.COMPLEX: 25,
}[complexity]
)
return await run_query(prompt, options)
# Uso
result = await smart_query("¿Cuántos TODOs hay?", TaskComplexity.SIMPLE)
result = await smart_query("Refactoriza el módulo de auth", TaskComplexity.MEDIUM)// TypeScript — escalado automático
type Model = "claude-haiku-3-5" | "claude-sonnet-4-5" | "claude-opus-4-5";
function selectModel(estimatedFiles: number): Model {
if (estimatedFiles <= 5) return "claude-haiku-3-5";
if (estimatedFiles <= 20) return "claude-sonnet-4-5";
return "claude-opus-4-5";
}
async function adaptiveQuery(
prompt: string,
projectFiles: number
): Promise<string> {
const options: ClaudeCodeOptions = {
model: selectModel(projectFiles),
max_turns: Math.min(5 + projectFiles * 2, 30),
};
for await (const message of query({ prompt, options })) {
if (message.type === "result") return message.result;
}
return "";
}Estimación de costo antes de ejecutar
# Python — estimación aproximada antes de ejecutar
def estimate_cost(
prompt_tokens: int,
estimated_output_tokens: int,
model: str
) -> float:
prices = {
"claude-haiku-3-5": (0.80, 4.0),
"claude-sonnet-4-5": (3.0, 15.0),
"claude-opus-4-5": (15.0, 75.0),
}
input_price, output_price = prices.get(model, (3.0, 15.0))
return (prompt_tokens * input_price + estimated_output_tokens * output_price) / 1_000_000
# Haiku para 10k tokens entrada, 1k salida
cost = estimate_cost(10_000, 1_000, "claude-haiku-3-5")
print(f"Estimación: ${cost:.4f}") # ~$0.0129. max_turns y el loop agentic
Qué es un “turn”
Un turn es un ciclo completo del loop agentic:
flowchart LR
A[Prompt inicial] --> B[AssistantMessage]
B --> C{¿Usa herramienta?}
C -->|Sí| D[ToolResultMessage]
D --> E[Siguiente AssistantMessage]
E --> F{¿Turn 2?}
F -->|Sigue| C
C -->|No| G[ResultMessage]
F -->|max_turns alcanzado| H[ResultMessage: error_max_turns]Cada vez que Claude usa una herramienta y recibe el resultado cuenta como un turn. Una query que usa 5 herramientas consecutivas usa 5 turns.
Qué pasa cuando se alcanza max_turns
# Python — detectar y manejar max_turns
from claude_code_sdk.types import ResultMessage
async def query_with_continuation(
prompt: str,
options: ClaudeCodeOptions,
max_continuation: int = 3
) -> str:
session_id = None
result = ""
for _ in range(max_continuation):
opts = options if session_id is None else ClaudeCodeOptions(
**{**vars(options), "resume": session_id}
)
async for message in query(prompt if session_id is None else "Continúa la tarea anterior", opts):
if isinstance(message, ResultMessage):
if message.subtype == "error_max_turns":
session_id = message.session_id
print(f"Max turns alcanzado, continuando sesión {session_id}")
break
else:
return message.result
return result// TypeScript — verificar turns usados
for await (const message of query({ prompt, options: { max_turns: 10 } })) {
if (message.type === "result") {
const efficiency = message.num_turns / 10;
if (efficiency > 0.8) {
console.warn(`Alta utilización de turns: ${message.num_turns}/10`);
console.warn("Considera aumentar max_turns o simplificar la tarea");
}
}
}Cómo estimar max_turns necesario
| Tipo de tarea | max_turns recomendado |
|---|---|
| Preguntas simples (sin tools) | 1-2 |
| Análisis de 1-5 archivos | 3-8 |
| Análisis de proyecto completo | 10-15 |
| Refactoring simple | 8-15 |
| Refactoring complejo | 15-30 |
| Migración de codebase | 25-50 |
# Python — helper para estimar turns según el prompt
import re
def estimate_max_turns(prompt: str, num_files: int = 0) -> int:
"""Estimación heurística de turns necesarios."""
base = 5
# Palabras clave que indican complejidad
complex_keywords = ["todos", "all", "complete", "full", "migrate", "refactor"]
simple_keywords = ["check", "count", "find", "list", "show"]
prompt_lower = prompt.lower()
if any(kw in prompt_lower for kw in complex_keywords):
base = 20
elif any(kw in prompt_lower for kw in simple_keywords):
base = 3
# Ajustar por número de archivos
file_factor = min(num_files // 5, 10)
return base + file_factor10. Ejemplos completos end-to-end
10.1 Agente analizador de código Python
Analiza un proyecto Python y genera un reporte de calidad.
# Python — agente_analizador.py
import asyncio
import json
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import ResultMessage
async def analyze_python_project(project_path: str) -> dict:
"""Analiza un proyecto Python y retorna métricas de calidad."""
system_prompt = """Eres un experto en calidad de código Python.
Analiza el proyecto y retorna un JSON con esta estructura exacta:
{
"files_analyzed": number,
"total_lines": number,
"issues": [
{"severity": "critical|high|medium|low", "file": "...", "line": N, "description": "..."}
],
"metrics": {
"avg_function_length": number,
"test_coverage_estimate": "high|medium|low|none",
"has_type_hints": boolean,
"has_docstrings": boolean
},
"summary": "string"
}
Responde SOLO con el JSON, sin texto adicional."""
options = ClaudeCodeOptions(
cwd=project_path,
system_prompt=system_prompt,
allowed_tools=["Read", "Glob", "Grep"],
model="claude-sonnet-4-5",
max_turns=15
)
prompt = """Analiza todos los archivos Python del proyecto:
1. Lee los archivos .py principales
2. Busca issues comunes (funciones muy largas, falta de tipos, etc.)
3. Retorna el JSON de análisis"""
result_text = ""
async for message in query(prompt, options):
if isinstance(message, ResultMessage):
if message.is_error:
raise RuntimeError(f"Análisis falló: {message.result}")
result_text = message.result
print(f"Análisis completado en {message.duration_ms}ms (${message.cost_usd:.4f})")
# Limpiar y parsear JSON
json_text = result_text.strip()
if json_text.startswith("```"):
json_text = "\n".join(json_text.split("\n")[1:-1])
return json.loads(json_text)
async def main():
report = await analyze_python_project("/home/user/my-project")
print(f"Archivos analizados: {report['files_analyzed']}")
print(f"Issues encontrados: {len(report['issues'])}")
critical = [i for i in report['issues'] if i['severity'] == 'critical']
if critical:
print(f"CRÍTICOS: {len(critical)}")
for issue in critical:
print(f" {issue['file']}:{issue['line']} - {issue['description']}")
asyncio.run(main())// TypeScript — analyzeProject.ts
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
interface ProjectAnalysis {
files_analyzed: number;
total_lines: number;
issues: Array<{
severity: "critical" | "high" | "medium" | "low";
file: string;
line: number;
description: string;
}>;
metrics: {
avg_function_length: number;
test_coverage_estimate: "high" | "medium" | "low" | "none";
has_type_hints: boolean;
has_docstrings: boolean;
};
summary: string;
}
async function analyzePythonProject(
projectPath: string
): Promise<ProjectAnalysis> {
const options: ClaudeCodeOptions = {
cwd: projectPath,
system_prompt: `Eres un experto en calidad de código Python.
Retorna SOLO un JSON válido con la estructura de análisis especificada.`,
allowed_tools: ["Read", "Glob", "Grep"],
model: "claude-sonnet-4-5",
max_turns: 15,
};
for await (const message of query({
prompt: "Analiza todos los archivos .py y retorna el JSON de métricas",
options,
})) {
if (message.type === "result" && !message.is_error) {
const jsonText = message.result.replace(/^```json?\n?|\n?```$/g, "");
return JSON.parse(jsonText) as ProjectAnalysis;
}
}
throw new Error("No se obtuvo resultado del análisis");
}
// Uso
const report = await analyzePythonProject("/home/user/my-project");
console.log(`Archivos: ${report.files_analyzed}`);
console.log(`Issues: ${report.issues.length}`);10.2 Agente generador de README (TypeScript)
// TypeScript — generateReadme.ts
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
import fs from "fs/promises";
import path from "path";
async function generateReadme(projectPath: string): Promise<void> {
const options: ClaudeCodeOptions = {
cwd: projectPath,
system_prompt: `Eres un technical writer experto.
Generas READMEs profesionales en Markdown con:
- Badge de versión, licencia, tests
- Descripción clara del proyecto
- Quick Start (menos de 5 pasos)
- API Reference si es librería
- Ejemplos de código reales del proyecto
- Contributing guide
Usa los archivos existentes para extraer información real, no inventes.`,
allowed_tools: ["Read", "Glob", "Grep"],
model: "claude-sonnet-4-5",
max_turns: 12,
permission_mode: "default",
};
console.log("Analizando proyecto para generar README...");
let readmeContent = "";
for await (const message of query({
prompt: `Genera un README.md completo y profesional para este proyecto.
Lee el package.json, los archivos principales y los tests para obtener información real.
Retorna SOLO el contenido del README en Markdown, sin comentarios adicionales.`,
options,
})) {
if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "text") {
process.stdout.write(".");
}
}
} else if (message.type === "result") {
if (message.is_error) {
throw new Error(`Error generando README: ${message.result}`);
}
readmeContent = message.result;
console.log(
`\nListo en ${message.duration_ms}ms ($${message.cost_usd.toFixed(4)})`
);
}
}
// Limpiar markdown wrapper si existe
if (readmeContent.startsWith("```markdown")) {
readmeContent = readmeContent.split("\n").slice(1, -1).join("\n");
}
const outputPath = path.join(projectPath, "README.md");
await fs.writeFile(outputPath, readmeContent, "utf-8");
console.log(`README guardado en ${outputPath}`);
}
generateReadme(process.argv[2] ?? process.cwd());10.3 Agente buscador de bugs con reporte
# Python — bug_hunter.py
import asyncio
from dataclasses import dataclass, field
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import AssistantMessage, ToolResultMessage, ResultMessage
@dataclass
class BugReport:
bugs: list[dict] = field(default_factory=list)
files_scanned: list[str] = field(default_factory=list)
tools_used: list[str] = field(default_factory=list)
cost_usd: float = 0.0
session_id: str = ""
async def hunt_bugs(project_path: str, language: str = "python") -> BugReport:
report = BugReport()
system = f"""Eres un experto en encontrar bugs en código {language}.
Tipos de bugs que buscas:
1. Null/None reference errors
2. Off-by-one errors en loops
3. Race conditions
4. Resource leaks (archivos no cerrados, conexiones no liberadas)
5. SQL injection / security vulnerabilities
6. Unhandled exceptions en puntos críticos
Para cada bug encontrado, incluye:
- Archivo y línea exacta
- Tipo de bug
- Impacto (crítico/alto/medio/bajo)
- Descripción técnica
- Fix recomendado con código"""
options = ClaudeCodeOptions(
cwd=project_path,
system_prompt=system,
allowed_tools=["Read", "Grep", "Glob"],
model="claude-opus-4-5", # Máxima capacidad para encontrar bugs sutiles
max_turns=20
)
async for message in query(
f"Encuentra todos los bugs en el código {language} de este proyecto. "
"Sé exhaustivo y revisa todos los archivos relevantes.", options
):
if isinstance(message, AssistantMessage):
for block in message.message.content:
if block.type == "tool_use":
report.tools_used.append(block.name)
if block.name == "Read" and "file_path" in block.input:
report.files_scanned.append(block.input["file_path"])
elif isinstance(message, ResultMessage):
report.cost_usd = message.cost_usd
report.session_id = message.session_id
# Parsear bugs del resultado
lines = message.result.split("\n")
current_bug = {}
for line in lines:
if line.startswith("### Bug") or line.startswith("## Bug"):
if current_bug:
report.bugs.append(current_bug)
current_bug = {"title": line.strip("# ")}
elif ":" in line and current_bug:
key, _, value = line.partition(":")
current_bug[key.strip().lower()] = value.strip()
if current_bug:
report.bugs.append(current_bug)
return report
async def main():
print("Iniciando análisis de bugs...")
report = await hunt_bugs("/home/user/vulnerable-app", "python")
print(f"\n=== REPORTE DE BUGS ===")
print(f"Archivos escaneados: {len(set(report.files_scanned))}")
print(f"Bugs encontrados: {len(report.bugs)}")
print(f"Herramientas usadas: {len(report.tools_used)} llamadas")
print(f"Costo del análisis: ${report.cost_usd:.4f}")
print(f"Sesión: {report.session_id}")
if report.bugs:
print("\nBugs críticos:")
critical = [b for b in report.bugs if "crítico" in str(b).lower() or "critical" in str(b).lower()]
for bug in critical[:5]:
print(f" - {bug.get('title', 'Sin título')}")
asyncio.run(main())Resumen del capítulo
En este capítulo cubriste los fundamentos completos de query():
- La función query() es un async generator que emite cuatro tipos de mensajes: SystemMessage, AssistantMessage, ToolResultMessage y ResultMessage
- ClaudeCodeOptions tiene más de 10 opciones que controlan herramientas, modelo, permisos, contexto y comportamiento
- Patrones de procesamiento: básico (solo resultado), streaming (tiempo real), auditoría (logging) y costo (monitoreo financiero)
- Manejo de errores robusto con retry y backoff exponencial para producción
- cwd es crítico para la seguridad — usa el directorio más específico posible
- system_prompt define el comportamiento del agente; mantenlo conciso
- max_turns debe estimarse según la complejidad de la tarea
En el siguiente capítulo aprenderás en detalle cada herramienta integrada y cómo combinarlas para casos de uso específicos.