Funciones

query()

La función async principal para interactuar con Claude Code. Devuelve un iterador async que produce mensajes a medida que llegan.
async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeCodeOptions | None = None
) -> AsyncIterator[Message]

Parámetros

ParámetroTipoDescripción
promptstr | AsyncIterable[dict]El prompt de entrada como string o iterable async para modo streaming
optionsClaudeCodeOptions | NoneObjeto de configuración opcional (por defecto ClaudeCodeOptions() si es None)

Retorna

Devuelve un AsyncIterator[Message] que produce mensajes de la conversación.

Ejemplo - Consulta simple

import asyncio
from claude_code_sdk import query

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

asyncio.run(main())

Ejemplo - Con opciones


import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    options = ClaudeCodeOptions(
        system_prompt="Eres un desarrollador experto en Python",
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Crea un servidor web en Python",
        options=options
    ):
        print(message)


asyncio.run(main())

tool()

Decorador para definir herramientas MCP con seguridad de tipos.
def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

Parámetros

ParámetroTipoDescripción
namestrIdentificador único para la herramienta
descriptionstrDescripción legible de lo que hace la herramienta
input_schematype | dict[str, Any]Esquema que define los parámetros de entrada de la herramienta (ver abajo)

Opciones de Esquema de Entrada

  1. Mapeo de tipo simple (recomendado):
    {"text": str, "count": int, "enabled": bool}
    
  2. Formato de esquema JSON (para validación compleja):
    {
        "type": "object",
        "properties": {
            "text": {"type": "string"},
            "count": {"type": "integer", "minimum": 0}
        },
        "required": ["text"]
    }
    

Retorna

Una función decoradora que envuelve la implementación de la herramienta y devuelve una instancia SdkMcpTool.

Ejemplo

from claude_code_sdk import tool
from typing import Any

@tool("greet", "Saludar a un usuario", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {
        "content": [{
            "type": "text",
            "text": f"¡Hola, {args['name']}!"
        }]
    }

create_sdk_mcp_server()

Crear un servidor MCP en proceso que se ejecuta dentro de tu aplicación Python.
def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

Parámetros

ParámetroTipoPor defectoDescripción
namestr-Identificador único para el servidor
versionstr"1.0.0"String de versión del servidor
toolslist[SdkMcpTool[Any]] | NoneNoneLista de funciones de herramientas creadas con el decorador @tool

Retorna

Devuelve un objeto McpSdkServerConfig que puede pasarse a ClaudeCodeOptions.mcp_servers.

Ejemplo

from claude_code_sdk import tool, create_sdk_mcp_server

@tool("add", "Sumar dos números", {"a": float, "b": float})
async def add(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Suma: {args['a'] + args['b']}"
        }]
    }

@tool("multiply", "Multiplicar dos números", {"a": float, "b": float})
async def multiply(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Producto: {args['a'] * args['b']}"
        }]
    }

calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]  # Pasar funciones decoradas
)

# Usar con Claude
options = ClaudeCodeOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

Clases

ClaudeSDKClient

Cliente para conversaciones bidireccionales e interactivas con Claude Code. Proporciona control completo sobre el flujo de conversación con soporte para streaming, interrupciones y envío dinámico de mensajes.
class ClaudeSDKClient:
    def __init__(self, options: ClaudeCodeOptions | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def disconnect(self) -> None

Métodos

MétodoDescripción
__init__(options)Inicializar el cliente con configuración opcional
connect(prompt)Conectar a Claude con un prompt inicial opcional o flujo de mensajes
query(prompt, session_id)Enviar una nueva solicitud en modo streaming
receive_messages()Recibir todos los mensajes de Claude como un iterador async
receive_response()Recibir mensajes hasta e incluyendo un ResultMessage
interrupt()Enviar señal de interrupción (solo funciona en modo streaming)
disconnect()Desconectar de Claude

Soporte de Context Manager

El cliente puede usarse como un context manager async para gestión automática de conexión:
async with ClaudeSDKClient() as client:
    await client.query("Hola Claude")
    async for message in client.receive_response():
        print(message)
Importante: Al iterar sobre mensajes, evita usar break para salir temprano ya que esto puede causar problemas de limpieza de asyncio. En su lugar, deja que la iteración se complete naturalmente o usa flags para rastrear cuando hayas encontrado lo que necesitas.

Ejemplo - Conversación interactiva

import asyncio
from claude_code_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        # Enviar mensaje inicial
        await client.query("Resolvamos un problema de matemáticas paso a paso")
        
        # Recibir y procesar respuesta
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Asistente: {block.text[:100]}...")
            elif isinstance(message, ResultMessage):
                print("Respuesta completa")
        
        # Enviar seguimiento basado en la respuesta
        await client.query("¿Cuál es el 15% de 80?")

asyncio.run(main())

Tipos

SdkMcpTool

Definición para una herramienta SDK MCP creada con el decorador @tool.
@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]
PropiedadTipoDescripción
namestrIdentificador único para la herramienta
descriptionstrDescripción legible
input_schematype[T] | dict[str, Any]Esquema para validación de entrada
handlerCallable[[T], Awaitable[dict[str, Any]]]Función async que maneja la ejecución de la herramienta

ClaudeCodeOptions

Dataclass de configuración para consultas de Claude Code.
@dataclass
class ClaudeCodeOptions:
    allowed_tools: list[str] = field(default_factory=list)
    max_thinking_tokens: int = 8000
    system_prompt: str | None = None
    append_system_prompt: str | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)
PropiedadTipoPor defectoDescripción
allowed_toolslist[str][]Lista de nombres de herramientas permitidas
max_thinking_tokensint8000Tokens máximos para el proceso de pensamiento
system_promptstr | NoneNoneReemplazar completamente el prompt del sistema por defecto
append_system_promptstr | NoneNoneTexto para agregar al prompt del sistema por defecto
mcp_serversdict[str, McpServerConfig] | str | Path{}Configuraciones de servidor MCP o ruta al archivo de configuración
permission_modePermissionMode | NoneNoneModo de permisos para uso de herramientas
continue_conversationboolFalseContinuar la conversación más reciente
resumestr | NoneNoneID de sesión para reanudar
max_turnsint | NoneNoneTurnos máximos de conversación
disallowed_toolslist[str][]Lista de nombres de herramientas no permitidas
modelstr | NoneNoneModelo de Claude a usar
permission_prompt_tool_namestr | NoneNoneNombre de herramienta MCP para prompts de permisos
cwdstr | Path | NoneNoneDirectorio de trabajo actual
settingsstr | NoneNoneRuta al archivo de configuración
add_dirslist[str | Path][]Directorios adicionales a los que Claude puede acceder
extra_argsdict[str, str | None]{}Argumentos CLI adicionales para pasar directamente al CLI
can_use_toolCanUseTool | NoneNoneFunción callback de permisos de herramientas
hooksdict[HookEvent, list[HookMatcher]] | NoneNoneConfiguraciones de hooks para interceptar eventos

PermissionMode

Modos de permisos para controlar la ejecución de herramientas.
PermissionMode = Literal[
    "default",           # Comportamiento de permisos estándar
    "acceptEdits",       # Auto-aceptar ediciones de archivos
    "plan",              # Modo planificación - sin ejecución
    "bypassPermissions"  # Omitir todas las verificaciones de permisos (usar con precaución)
]

McpSdkServerConfig

Configuración para servidores SDK MCP creados con create_sdk_mcp_server().
class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # Instancia del servidor MCP

McpServerConfig

Tipo unión para configuraciones de servidor MCP.
McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

McpStdioServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # Opcional para compatibilidad hacia atrás
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpSSEServerConfig

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

McpHttpServerConfig

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

Tipos de Mensaje

Message

Tipo unión de todos los mensajes posibles.
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

UserMessage

Mensaje de entrada del usuario.
@dataclass
class UserMessage:
    content: str | list[ContentBlock]

AssistantMessage

Mensaje de respuesta del asistente con bloques de contenido.
@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

SystemMessage

Mensaje del sistema con metadatos.
@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

ResultMessage

Mensaje de resultado final con información de costo y uso.
@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None

Tipos de Bloque de Contenido

ContentBlock

Tipo unión de todos los bloques de contenido.
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Bloque de contenido de texto.
@dataclass
class TextBlock:
    text: str

ThinkingBlock

Bloque de contenido de pensamiento (para modelos con capacidad de pensamiento).
@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

Bloque de solicitud de uso de herramienta.
@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

ToolResultBlock

Bloque de resultado de ejecución de herramienta.
@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

Tipos de Error

ClaudeSDKError

Clase de excepción base para todos los errores del SDK.
class ClaudeSDKError(Exception):
    """Error base para Claude SDK."""

CLINotFoundError

Se lanza cuando el CLI de Claude Code no está instalado o no se encuentra.
class CLINotFoundError(CLIConnectionError):
    def __init__(self, message: str = "Claude Code not found", cli_path: str | None = None):
        """
        Args:
            message: Mensaje de error (por defecto: "Claude Code not found")
            cli_path: Ruta opcional al CLI que no se encontró
        """

CLIConnectionError

Se lanza cuando falla la conexión a Claude Code.
class CLIConnectionError(ClaudeSDKError):
    """Falló la conexión a Claude Code."""

ProcessError

Se lanza cuando falla el proceso de Claude Code.
class ProcessError(ClaudeSDKError):
    def __init__(self, message: str, exit_code: int | None = None, stderr: str | None = None):
        self.exit_code = exit_code
        self.stderr = stderr

CLIJSONDecodeError

Se lanza cuando falla el análisis JSON.
class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: La línea que falló al analizar
            original_error: La excepción original de decodificación JSON
        """
        self.line = line
        self.original_error = original_error

Tipos de Hook

HookEvent

Tipos de eventos de hook soportados. Nota que debido a limitaciones de configuración, el SDK de Python no soporta hooks SessionStart, SessionEnd y Notification.
HookEvent = Literal[
    "PreToolUse",      # Llamado antes de la ejecución de herramientas
    "PostToolUse",     # Llamado después de la ejecución de herramientas
    "UserPromptSubmit", # Llamado cuando el usuario envía un prompt
    "Stop",            # Llamado al detener la ejecución
    "SubagentStop",    # Llamado cuando un subagente se detiene
    "PreCompact"       # Llamado antes de la compactación de mensajes
]

HookCallback

Definición de tipo para funciones callback de hook.
HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]
Parámetros:
  • input_data: Datos de entrada específicos del hook (ver documentación de hooks)
  • tool_use_id: Identificador opcional de uso de herramienta (para hooks relacionados con herramientas)
  • context: Contexto del hook con información adicional
Devuelve un diccionario que puede contener:
  • decision: "block" para bloquear la acción
  • systemMessage: Mensaje del sistema para agregar a la transcripción
  • hookSpecificOutput: Datos de salida específicos del hook

HookContext

Información de contexto pasada a los callbacks de hook.
@dataclass
class HookContext:
    signal: Any | None = None  # Futuro: soporte de señal de aborto

HookMatcher

Configuración para hacer coincidir hooks con eventos o herramientas específicas.
@dataclass
class HookMatcher:
    matcher: str | None = None        # Nombre de herramienta o patrón a coincidir (ej., "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)  # Lista de callbacks a ejecutar

Ejemplo de Uso de Hook

from claude_code_sdk import query, ClaudeCodeOptions, HookMatcher, HookContext
from typing import Any

async def validate_bash_command(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Validar y potencialmente bloquear comandos bash peligrosos."""
    if input_data['tool_name'] == 'Bash':
        command = input_data['tool_input'].get('command', '')
        if 'rm -rf /' in command:
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': 'Comando peligroso bloqueado'
                }
            }
    return {}

async def log_tool_use(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Registrar todo el uso de herramientas para auditoría."""
    print(f"Herramienta usada: {input_data.get('tool_name')}")
    return {}

options = ClaudeCodeOptions(
    hooks={
        'PreToolUse': [
            HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
            HookMatcher(hooks=[log_tool_use])  # Se aplica a todas las herramientas
        ],
        'PostToolUse': [
            HookMatcher(hooks=[log_tool_use])
        ]
    }
)

async for message in query(
    prompt="Analiza esta base de código",
    options=options
):
    print(message)

Tipos de Entrada/Salida de Herramientas

Documentación de esquemas de entrada/salida para todas las herramientas integradas de Claude Code. Aunque el SDK de Python no exporta estos como tipos, representan la estructura de entradas y salidas de herramientas en mensajes.

Task

Nombre de herramienta: Task Entrada:
{
    "description": str,      # Una descripción corta (3-5 palabras) de la tarea
    "prompt": str,           # La tarea para que el agente realice
    "subagent_type": str     # El tipo de agente especializado a usar
}
Salida:
{
    "result": str,                    # Resultado final del subagente
    "usage": dict | None,             # Estadísticas de uso de tokens
    "total_cost_usd": float | None,  # Costo total en USD
    "duration_ms": int | None         # Duración de ejecución en milisegundos
}

Bash

Nombre de herramienta: Bash Entrada:
{
    "command": str,                  # El comando a ejecutar
    "timeout": int | None,           # Timeout opcional en milisegundos (máx 600000)
    "description": str | None,       # Descripción clara y concisa (5-10 palabras)
    "run_in_background": bool | None # Establecer en true para ejecutar en segundo plano
}
Salida:
{
    "output": str,              # Salida combinada de stdout y stderr
    "exitCode": int,            # Código de salida del comando
    "killed": bool | None,      # Si el comando fue terminado por timeout
    "shellId": str | None       # ID de shell para procesos en segundo plano
}

Edit

Nombre de herramienta: Edit Entrada:
{
    "file_path": str,           # La ruta absoluta al archivo a modificar
    "old_string": str,          # El texto a reemplazar
    "new_string": str,          # El texto con el que reemplazarlo
    "replace_all": bool | None  # Reemplazar todas las ocurrencias (por defecto False)
}
Salida:
{
    "message": str,      # Mensaje de confirmación
    "replacements": int, # Número de reemplazos realizados
    "file_path": str     # Ruta del archivo que fue editado
}

MultiEdit

Nombre de herramienta: MultiEdit Entrada:
{
    "file_path": str,     # La ruta absoluta al archivo a modificar
    "edits": [            # Array de operaciones de edición
        {
            "old_string": str,          # El texto a reemplazar
            "new_string": str,          # El texto con el que reemplazarlo
            "replace_all": bool | None  # Reemplazar todas las ocurrencias
        }
    ]
}
Salida:
{
    "message": str,       # Mensaje de éxito
    "edits_applied": int, # Número total de ediciones aplicadas
    "file_path": str      # Ruta del archivo que fue editado
}

Read

Nombre de herramienta: Read Entrada:
{
    "file_path": str,       # La ruta absoluta al archivo a leer
    "offset": int | None,   # El número de línea desde donde empezar a leer
    "limit": int | None     # El número de líneas a leer
}
Salida (Archivos de texto):
{
    "content": str,         # Contenido del archivo con números de línea
    "total_lines": int,     # Número total de líneas en el archivo
    "lines_returned": int   # Líneas realmente devueltas
}
Salida (Imágenes):
{
    "image": str,       # Datos de imagen codificados en base64
    "mime_type": str,   # Tipo MIME de la imagen
    "file_size": int    # Tamaño del archivo en bytes
}

Write

Nombre de herramienta: Write Entrada:
{
    "file_path": str,  # La ruta absoluta al archivo a escribir
    "content": str     # El contenido a escribir en el archivo
}
Salida:
{
    "message": str,        # Mensaje de éxito
    "bytes_written": int,  # Número de bytes escritos
    "file_path": str       # Ruta del archivo que fue escrito
}

Glob

Nombre de herramienta: Glob Entrada:
{
    "pattern": str,       # El patrón glob para coincidir archivos
    "path": str | None    # El directorio donde buscar (por defecto cwd)
}
Salida:
{
    "matches": list[str],  # Array de rutas de archivos coincidentes
    "count": int,          # Número de coincidencias encontradas
    "search_path": str     # Directorio de búsqueda usado
}

Grep

Nombre de herramienta: Grep Entrada:
{
    "pattern": str,                    # El patrón de expresión regular
    "path": str | None,                # Archivo o directorio donde buscar
    "glob": str | None,                # Patrón glob para filtrar archivos
    "type": str | None,                # Tipo de archivo a buscar
    "output_mode": str | None,         # "content", "files_with_matches", o "count"
    "-i": bool | None,                 # Búsqueda insensible a mayúsculas
    "-n": bool | None,                 # Mostrar números de línea
    "-B": int | None,                  # Líneas a mostrar antes de cada coincidencia
    "-A": int | None,                  # Líneas a mostrar después de cada coincidencia
    "-C": int | None,                  # Líneas a mostrar antes y después
    "head_limit": int | None,          # Limitar salida a las primeras N líneas/entradas
    "multiline": bool | None           # Habilitar modo multilínea
}
Salida (modo content):
{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None
        }
    ],
    "total_matches": int
}
Salida (modo files_with_matches):
{
    "files": list[str],  # Archivos que contienen coincidencias
    "count": int         # Número de archivos con coincidencias
}

NotebookEdit

Nombre de herramienta: NotebookEdit Entrada:
{
    "notebook_path": str,                     # Ruta absoluta al notebook de Jupyter
    "cell_id": str | None,                    # El ID de la celda a editar
    "new_source": str,                        # La nueva fuente para la celda
    "cell_type": "code" | "markdown" | None,  # El tipo de la celda
    "edit_mode": "replace" | "insert" | "delete" | None  # Tipo de operación de edición
}
Salida:
{
    "message": str, # Mensaje de éxito
    "edit_type": "replaced" | "inserted" | "deleted",  # Tipo de edición realizada
    "cell_id": str | None,                       # ID de celda que fue afectada
    "total_cells": int                           # Total de celdas en el notebook después de la edición
}

WebFetch

Nombre de herramienta: WebFetch Entrada:
{
    "url": str,     # La URL de la cual obtener contenido
    "prompt": str   # El prompt a ejecutar en el contenido obtenido
}
Salida:
{
    "response": str,           # Respuesta del modelo de IA al prompt
    "url": str,                # URL que fue obtenida
    "final_url": str | None,   # URL final después de redirecciones
    "status_code": int | None  # Código de estado HTTP
}

WebSearch

Nombre de herramienta: WebSearch Entrada:
{
    "query": str,                        # La consulta de búsqueda a usar
    "allowed_domains": list[str] | None, # Solo incluir resultados de estos dominios
    "blocked_domains": list[str] | None  # Nunca incluir resultados de estos dominios
}
Salida:
{
    "results": [
        {
            "title": str,
            "url": str,
            "snippet": str,
            "metadata": dict | None
        }
    ],
    "total_results": int,
    "query": str
}

TodoWrite

Nombre de herramienta: TodoWrite Entrada:
{
    "todos": [
        {
            "content": str, # La descripción de la tarea
            "status": "pending" | "in_progress" | "completed",  # Estado de la tarea
            "activeForm": str                            # Forma activa de la descripción
        }
    ]
}
Salida:
{
    "message": str,  # Mensaje de éxito
    "stats": {
        "total": int,
        "pending": int,
        "in_progress": int,
        "completed": int
    }
}

BashOutput

Nombre de herramienta: BashOutput Entrada:
{
    "bash_id": str,       # El ID del shell en segundo plano
    "filter": str | None  # Regex opcional para filtrar líneas de salida
}
Salida:
{
    "output": str, # Nueva salida desde la última verificación
    "status": "running" | "completed" | "failed",       # Estado actual del shell
    "exitCode": int | None # Código de salida cuando se completa
}

KillBash

Nombre de herramienta: KillBash Entrada:
{
    "shell_id": str  # El ID del shell en segundo plano a terminar
}
Salida:
{
    "message": str,  # Mensaje de éxito
    "shell_id": str  # ID del shell terminado
}

ExitPlanMode

Nombre de herramienta: ExitPlanMode Entrada:
{
    "plan": str  # El plan a ejecutar por el usuario para aprobación
}
Salida:
{
    "message": str,          # Mensaje de confirmación
    "approved": bool | None  # Si el usuario aprobó el plan
}

ListMcpResources

Nombre de herramienta: ListMcpResources Entrada:
{
    "server": str | None  # Nombre de servidor opcional para filtrar recursos
}
Salida:
{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str
        }
    ],
    "total": int
}

ReadMcpResource

Nombre de herramienta: ReadMcpResource Entrada:
{
    "server": str,  # El nombre del servidor MCP
    "uri": str      # La URI del recurso a leer
}
Salida:
{
    "contents": [
        {
            "uri": str,
            "mimeType": str | None,
            "text": str | None,
            "blob": str | None
        }
    ],
    "server": str
}

Ejemplo de Uso

Operaciones básicas de archivos

from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, ToolUseBlock
import asyncio

async def create_project():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )
    
    async for message in query(
        prompt="Crea una estructura de proyecto Python con setup.py",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Usando herramienta: {block.name}")

asyncio.run(create_project())

Manejo de errores

from claude_code_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

try:
    async for message in query(prompt="Hola"):
        print(message)
except CLINotFoundError:
    print("Por favor instala Claude Code: npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"El proceso falló con código de salida: {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Falló al analizar respuesta: {e}")

Modo streaming con cliente

from claude_code_sdk import ClaudeSDKClient
import asyncio

async def interactive_session():
    async with ClaudeSDKClient() as client:
        # Enviar mensaje inicial
        await client.query("¿Cómo está el clima?")
        
        # Procesar respuestas
        async for msg in client.receive_response():
            print(msg)
        
        # Enviar seguimiento
        await client.query("Cuéntame más sobre eso")
        
        # Procesar respuesta de seguimiento
        async for msg in client.receive_response():
            print(msg)

asyncio.run(interactive_session())

Usando herramientas personalizadas

from claude_code_sdk import (
    query,
    ClaudeCodeOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)
import asyncio
from typing import Any

# Definir herramientas personalizadas con el decorador @tool
@tool("calculate", "Realizar cálculos matemáticos", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {
            "content": [{
                "type": "text",
                "text": f"Resultado: {result}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"Error: {str(e)}"
            }],
            "is_error": True
        }

@tool("get_time", "Obtener hora actual", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "content": [{
            "type": "text",
            "text": f"Hora actual: {current_time}"
        }]
    }

async def main():
    # Crear servidor SDK MCP con herramientas personalizadas
    my_server = create_sdk_mcp_server(
        name="utilities",
        version="1.0.0",
        tools=[calculate, get_time]
    )
    
    # Configurar opciones con el servidor
    options = ClaudeCodeOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=[
            "mcp__utils__calculate",
            "mcp__utils__get_time"
        ]
    )
    
    # Consultar Claude con herramientas personalizadas disponibles
    async for message in query(
        prompt="¿Cuánto es 123 * 456 y qué hora es?",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

Ver también