Capítulo 3: Query Básico

La Función query()

query() es la API principal del SDK. Es asíncrona y retorna un stream de mensajes.

Ejemplo Mínimo

import asyncio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="¿Cuánto es 2 + 2?"):
        print(message)

asyncio.run(main())

TypeScript

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

for await (const message of query({
  prompt: "¿Cuánto es 2 + 2?"
})) {
  console.log(message);
}

Tipos de Mensajes

El stream retorna diferentes tipos de mensajes:

flowchart TB
    subgraph Mensajes["TIPOS DE MENSAJES"]
        AM["AssistantMessage<br/>Mensaje del agente con texto o tool use"]
        UM["UserMessage<br/>Mensaje del usuario en conversaciones"]
        SM["SystemMessage<br/>Mensaje del sistema (init, status)"]
        RM["ResultMessage<br/>Resultado final de la operación"]
    end

Filtrar por Tipo

from claude_agent_sdk import (
    query,
    AssistantMessage,
    ResultMessage,
    TextBlock
)

async def main():
    async for msg in query(prompt="Explica qué es Python"):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)
        elif isinstance(msg, ResultMessage):
            print(f"\n--- Resultado: {msg.result}")

asyncio.run(main())

ClaudeAgentOptions

Configura el comportamiento del agente:

from claude_agent_sdk import query, ClaudeAgentOptions

options = ClaudeAgentOptions(
    system_prompt="Eres un asistente de Python experto",
    max_turns=5,                    # Máximo de turnos
    allowed_tools=["Read", "Bash"], # Herramientas permitidas
    cwd="/mi/proyecto"              # Directorio de trabajo
)

async for msg in query(prompt="Lista los archivos .py", options=options):
    print(msg)

Opciones Principales

Opción	Tipo	Descripción
system_prompt	str	Prompt del sistema
max_turns	int	Límite de turnos
allowed_tools	list	Herramientas habilitadas
cwd	str/Path	Directorio de trabajo
permission_mode	str	Modo de permisos
mcp_servers	dict	Servidores MCP
hooks	dict	Hooks personalizados

Modos de Permisos

options = ClaudeAgentOptions(
    # Acepta automáticamente ediciones de archivos
    permission_mode="acceptEdits"
)

options = ClaudeAgentOptions(
    # Bypasa todos los permisos (cuidado!)
    permission_mode="bypassPermissions"
)

ClaudeSDKClient para Conversaciones

Para conversaciones bidireccionales:

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],
    permission_mode="acceptEdits"
)

async with ClaudeSDKClient(options=options) as client:
    # Primera consulta
    await client.query("Crea un archivo hello.py")
    async for msg in client.receive_response():
        print(msg)

    # Segunda consulta (mantiene contexto)
    await client.query("Ahora ejecútalo")
    async for msg in client.receive_response():
        print(msg)

Manejo de Errores

from claude_agent_sdk import (
    query,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError
)

async def safe_query(prompt: str):
    try:
        async for msg in query(prompt=prompt):
            print(msg)
    except CLINotFoundError:
        print("Error: Claude Code no está instalado")
    except CLIConnectionError:
        print("Error: No hay conexión")
    except ProcessError as e:
        print(f"Error: Proceso falló con código {e.exit_code}")
    except CLIJSONDecodeError:
        print("Error: Respuesta inválida")
    except ClaudeSDKError as e:
        print(f"Error general: {e}")

Ejemplo Completo: Analizador de Código

import asyncio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock
)

async def analizar_codigo(ruta: str):
    options = ClaudeAgentOptions(
        system_prompt="""Eres un experto en análisis de código.
        Analiza el código y reporta:
        - Complejidad
        - Posibles bugs
        - Sugerencias de mejora""",
        allowed_tools=["Read", "Glob", "Grep"],
        cwd=ruta
    )

    prompt = "Analiza los archivos Python en este proyecto"

    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(analizar_codigo("/mi/proyecto"))

Streaming vs Resultado Final

Solo Resultado Final

async def solo_resultado():
    resultado = None
    async for msg in query(prompt="¿Qué hora es?"):
        if hasattr(msg, 'result'):
            resultado = msg.result
    return resultado

Streaming Completo

async def con_streaming():
    async for msg in query(prompt="Cuenta del 1 al 10"):
        # Procesa cada mensaje conforme llega
        print(msg, end='', flush=True)

Resumen

• query() es la función principal para interactuar con el agente
• Retorna un stream asíncrono de mensajes
• ClaudeAgentOptions configura el comportamiento
• ClaudeSDKClient permite conversaciones bidireccionales
• Siempre maneja errores apropiadamente