Conceptos básicos de orquestación agentiva

10 ejemplos para empezar con la orquestación agentiva: 7 básicos y 3 intermedios.

Prerrequisitos

Instala el paquete Python: pip install claude-agent-sdk.
Configura una clave API en tu entorno: export ANTHROPIC_API_KEY=sk-ant-....
Estos ejemplos asumen que ya conoces la estructura de una única llamada query(); si no es así, empieza primero con los conceptos básicos del SDK de Claude Agent.
"Orquestación" aquí significa código de aplicación (o un agente de nivel superior) que decide cuándo llamar a subagentes y cómo combinar sus resultados, no ninguna característica individual del SDK en sí.

Ejemplos básicos

1. Un orquestador mínimo: Una tarea, un subagente

El orquestador más pequeño posible: delega una tarea a un único subagente y lee su resultado.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        allowed_tools=["file_edit"],
        subagents=[
            SubagentConfig(
                name="changelog-writer",
                description="Escribe una entrada en CHANGELOG.md a partir de un resumen de diff.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Usa el subagente changelog-writer para añadir una entrada en CHANGELOG.md para este lanzamiento.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Resultado: {message['result']}")
 
asyncio.run(main())

El orquestador aquí es una única llamada query() de nivel superior; el propio modelo decide invocar al único subagente que tiene.
Este es todo el patrón: un agente orquestador, un trabajador con su propio ámbito y un resultado que se devuelve.
Nada de esto necesita ser más complejo hasta que surja una necesidad real de descomposición o especialización.

Relacionado: Bucles de agente único vs. sistemas multiagente: Un modelo mental - cuándo vale la pena recurrir a este patrón

2. Combinación de resultados de dos subagentes

Despacha dos subagentes y combina sus resultados en una única salida.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Resume los cambios del frontend en un PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Resume los cambios del backend en un PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Usa ambos subagentes de resumen, luego escribe una descripción combinada del PR.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

El modelo orquestador decide llamar a ambos subagentes y es responsable de combinar sus resultados en una única descripción.
Cada subagente solo ve su propia parte de la tarea, no la salida del otro, a menos que el orquestador se la pase explícitamente.
Esta es la semilla de cada patrón de abanico/fusión cubierto en profundidad más adelante en esta sección.

3. Una cadena de prompts fija

Ejecuta una secuencia fija de pasos, cada uno dependiendo del anterior.

import asyncio
from claude_agent_sdk import query, AgentOptions
 
async def summarize_then_translate(text: str, language: str) -> str:
    summary_chunks = []
    async for message in query(prompt=f"Resume en tres viñetas:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Traduce esto a {language}, mantenlo en tres viñetas:\n\n{summary}"
    async for message in query(prompt=prompt):
        if message.get("type") == "text":
            translated_chunks.append(message["text"])
    return "".join(translated_chunks)
 
print(asyncio.run(summarize_then_translate("...", "Español")))

Cada paso es una llamada query() separada cuyo prompt se construye a partir de la salida del paso anterior.
No hay una decisión dinámica aquí sobre qué paso ejecutar a continuación; la secuencia está fijada por el código de llamada.
Esta es una cadena de prompts, el patrón de orquestación más simple, y a menudo todo lo que una tarea necesita cuando sus pasos son siempre los mismos y siempre en el mismo orden.

Relacionado: Cadena de prompts vs. Enrutamiento: Eligiendo tu patrón de orquestación - cuándo una cadena fija es la decisión correcta

4. Un enrutador simple entre dos caminos

Elige qué subagente invocar basándose en la tarea entrante, en lugar de ejecutar siempre la misma secuencia.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_ticket(ticket_text: str) -> None:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="bug-triager",
                description="Clasifica los informes de errores: reproduce, etiqueta la gravedad.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Define el alcance de las solicitudes de características en un plan de implementación aproximado.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Lee este ticket, decide si es un error o una solicitud de característica, "
        f"y enrútalo al subagente correspondiente:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("El botón de inicio de sesión no hace nada en Safari 18."))

La decisión de enrutamiento, qué subagente maneja este ticket, es tomada por el modelo en tiempo de ejecución, no fijada en el código de llamada.
Ambos destinos están disponibles en cada llamada; solo uno (a veces más) se invoca realmente, basándose en el contenido del ticket.
El enrutamiento se adapta a entradas variables; una cadena fija obligaría a cada ticket a pasar por los mismos pasos independientemente del tipo.

5. Despacho de subagentes independientes en paralelo

Ejecuta varias subtareas de forma similar concurrentemente en lugar de una tras otra.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def audit_packages(packages: list[str]) -> None:
    subagents = [
        SubagentConfig(
            name=f"audit-{pkg}",
            description=f"Comprueba el paquete {pkg} en busca de dependencias obsoletas.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Ejecuta todos los subagentes de auditoría para {packages} y lista los hallazgos por paquete."
 
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            print(f"[{message['subagent_name']}] {message['result']}")
 
asyncio.run(audit_packages(["billing", "auth", "search"]))

Un subagente por paquete significa que cada auditoría se ejecuta en su propio contexto, sin estado compartido entre ellos.
Dado que los paquetes no están relacionados, el bucle subyacente puede despachar estas invocaciones de subagente concurrentemente en lugar de serialmente.
Este es el patrón al que recurrir siempre que una tarea sea "haz lo mismo N veces, sobre N entradas independientes".

Relacionado: Creación de subagentes para investigación y delegación paralela - este patrón en profundidad

6. Limitar la profundidad de delegación

Evita que un subagente genere sus propios subagentes más allá de un nivel.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Investiga un tema; no puede delegar más.",
    allowed_tools=["web_search"],
    # No hay campo `subagents` en el propio worker: no tiene capacidad de delegación
    # propia, por lo que la profundidad se limita a un nivel aquí.
)
 
options = AgentOptions(subagents=[worker])

El orquestador puede delegar a research-worker, pero research-worker no tiene subagentes propios a los que delegar.
Limitar la profundidad de esta manera es una elección de diseño deliberada, no un valor predeterminado del SDK; nada te impide anidar subagentes más si lo configuras.
La profundidad sin límites hace que los costos y los modos de fallo sean mucho más difíciles de razonar a medida que un sistema crece.

Relacionado: Barreras de seguridad para sistemas multiagente: Limitación de costos y ámbito - profundidad, acceso a herramientas y límites de gasto combinados

7. Reintentar una llamada fallida a un subagente

Envuelve la invocación de un subagente con un reintento básico en lugar de fallar toda la tarea en un intento fallido.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def run_with_retry(prompt: str, options: AgentOptions, attempts: int = 3):
    last_error = None
    for attempt in range(attempts):
        try:
            results = []
            async for message in query(prompt=prompt, options=options):
                if message.get("type") == "subagent_result":
                    results.append(message["result"])
            return results
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(2 ** attempt)  # retroceso antes de reintentar
    raise RuntimeError(f"Falló después de {attempts} intentos") from last_error

Una llamada fallida a un subagente se trata como cualquier otro fallo que se pueda reintentar, no como un caso especial que el orquestador ignora.
El retroceso exponencial entre intentos evita sobrecargar una dependencia inestable inmediatamente después de que falle.
Los sistemas reales también necesitan un límite en el número total de intentos y una solución alternativa para cuando se agotan los reintentos, cubierto en el artículo de recuperación de errores.

Relacionado: Estrategias de recuperación de errores y reintentos en bucles de agentes - reintentos, soluciones alternativas y disyuntores

Ejemplos intermedios

8. Enrutamiento más despacho en abanico paralelo

Combina una decisión de enrutamiento con un despacho paralelo una vez que se elige el camino.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Cambio pequeño: enruta directamente a un revisor, no se necesita abanico.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Revisa el cambio de un solo archivo.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Usa el subagente reviewer en {changed_files[0]}."
    else:
        # Cambio mayor: abanico de un revisor por archivo, en paralelo.
        options = AgentOptions(
            subagents=[
                SubagentConfig(
                    name=f"reviewer-{i}",
                    description=f"Revisa {f}.",
                    allowed_tools=["file_edit"],
                )
                for i, f in enumerate(changed_files)
            ]
        )
        prompt = f"Usa todos los subagentes de revisión para revisar {changed_files} en paralelo."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

El paso de enrutamiento decide la forma del trabajo (un revisor o muchos) antes de que se ejecute cualquier subagente.
Una vez enrutado al camino de abanico, las subtareas son independientes y se despachan de la misma manera que el ejemplo de auditoría paralela.
La mezcla de enrutamiento y abanico es común en la práctica: el enrutamiento elige la estrategia, el abanico la ejecuta.

9. Un disyuntor alrededor de una herramienta inestable

Deja de reintentar un subagente cuya herramienta subyacente ha fallado repetidamente, en lugar de reintentar indefinidamente.

import asyncio
import time
from claude_agent_sdk import query, AgentOptions
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_after: float = 60.0):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_after = reset_after
        self.opened_at: float | None = None
 
    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.monotonic() - self.opened_at > self.reset_after:
            self.opened_at = None  # semiabierto: permite un intento
            self.failures = 0
            return False
        return True
 
    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()
 
    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None
 
breaker = CircuitBreaker()
 
async def call_flaky_subagent(prompt: str, options: AgentOptions):
    if breaker.is_open():
        raise RuntimeError("Circuito abierto: omitiendo llamada, la dependencia parecía poco saludable")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

El disyuntor rastrea los fallos consecutivos y deja de emitir nuevas llamadas una vez que se cruza un umbral, en lugar de reintentar una dependencia que claramente está caída.
Después de reset_after segundos, permite pasar exactamente un intento (semiabierto) para probar si la dependencia se ha recuperado.
Esto protege al resto del orquestador de gastar tiempo y tokens en llamadas que probablemente volverán a fallar inmediatamente.

10. Puntuación de la salida del orquestador antes de aceptarla

Añade una comprobación ligera que rechaza un resultado combinado obviamente malo en lugar de devolverlo sin verificar.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
def looks_reasonable(result: str) -> bool:
    return len(result.strip()) > 20 and "TODO" not in result
 
async def orchestrate_with_check(prompt: str, options: AgentOptions) -> str:
    final_result = ""
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            final_result = message["result"]
 
    if not looks_reasonable(final_result):
        raise ValueError("El resultado del orquestador falló la comprobación de cordura")
    return final_result

looks_reasonable es un sustituto de cualquier comprobación determinista y ligera que tenga sentido para tu tarea: longitud, campos requeridos, ausencia de texto de marcador de posición.
Comprobar la salida antes de que llegue a un llamador detecta una clase de fallos que un reintento por sí solo no haría, ya que la llamada en sí tuvo éxito, el contenido simplemente estaba mal.
Esta es una vista previa ligera de la disciplina de evaluación cubierta en su totalidad en la lista de verificación de preparación para producción.

Relacionado: Evaluación de la calidad del agente: Una lista de verificación para la preparación de producción - puntuación de la tasa de éxito, el costo y los modos de fallo de forma sistemática

Versiones de la pila: Escrito contra la línea de modelos Claude actual a partir de ~junio de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado), y Claude Haiku 4.5 - y el SDK de Claude Agent (última versión). Los nombres de los modelos, las versiones del SDK y los precios cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.

Conceptos básicos de orquestación agentiva

10 ejemplos para empezar con la orquestación agentiva: 7 básicos y 3 intermedios.

Prerrequisitos

Instala el paquete Python: pip install claude-agent-sdk.
Configura una clave API en tu entorno: export ANTHROPIC_API_KEY=sk-ant-....
Estos ejemplos asumen que ya conoces la estructura de una única llamada query(); si no es así, empieza primero con los conceptos básicos del SDK de Claude Agent.
"Orquestación" aquí significa código de aplicación (o un agente de nivel superior) que decide cuándo llamar a subagentes y cómo combinar sus resultados, no ninguna característica individual del SDK en sí.

Ejemplos básicos

1. Un orquestador mínimo: Una tarea, un subagente

El orquestador más pequeño posible: delega una tarea a un único subagente y lee su resultado.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        allowed_tools=["file_edit"],
        subagents=[
            SubagentConfig(
                name="changelog-writer",
                description="Escribe una entrada en CHANGELOG.md a partir de un resumen de diff.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Usa el subagente changelog-writer para añadir una entrada en CHANGELOG.md para este lanzamiento.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Resultado: {message['result']}")
 
asyncio.run(main())

El orquestador aquí es una única llamada query() de nivel superior; el propio modelo decide invocar al único subagente que tiene.
Este es todo el patrón: un agente orquestador, un trabajador con su propio ámbito y un resultado que se devuelve.
Nada de esto necesita ser más complejo hasta que surja una necesidad real de descomposición o especialización.

Relacionado: Bucles de agente único vs. sistemas multiagente: Un modelo mental - cuándo vale la pena recurrir a este patrón

2. Combinación de resultados de dos subagentes

Despacha dos subagentes y combina sus resultados en una única salida.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Resume los cambios del frontend en un PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Resume los cambios del backend en un PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Usa ambos subagentes de resumen, luego escribe una descripción combinada del PR.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

El modelo orquestador decide llamar a ambos subagentes y es responsable de combinar sus resultados en una única descripción.
Cada subagente solo ve su propia parte de la tarea, no la salida del otro, a menos que el orquestador se la pase explícitamente.
Esta es la semilla de cada patrón de abanico/fusión cubierto en profundidad más adelante en esta sección.

3. Una cadena de prompts fija

Ejecuta una secuencia fija de pasos, cada uno dependiendo del anterior.

import asyncio
from claude_agent_sdk import query, AgentOptions
 
async def summarize_then_translate(text: str, language: str) -> str:
    summary_chunks = []
    async for message in query(prompt=f"Resume en tres viñetas:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Traduce esto a {language}, mantenlo en tres viñetas:\n\n{summary}"
    async for message in query(prompt=prompt):
        if message.get("type") == "text":
            translated_chunks.append(message["text"])
    return "".join(translated_chunks)
 
print(asyncio.run(summarize_then_translate("...", "Español")))

Cada paso es una llamada query() separada cuyo prompt se construye a partir de la salida del paso anterior.
No hay una decisión dinámica aquí sobre qué paso ejecutar a continuación; la secuencia está fijada por el código de llamada.
Esta es una cadena de prompts, el patrón de orquestación más simple, y a menudo todo lo que una tarea necesita cuando sus pasos son siempre los mismos y siempre en el mismo orden.

Relacionado: Cadena de prompts vs. Enrutamiento: Eligiendo tu patrón de orquestación - cuándo una cadena fija es la decisión correcta

4. Un enrutador simple entre dos caminos

Elige qué subagente invocar basándose en la tarea entrante, en lugar de ejecutar siempre la misma secuencia.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_ticket(ticket_text: str) -> None:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="bug-triager",
                description="Clasifica los informes de errores: reproduce, etiqueta la gravedad.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Define el alcance de las solicitudes de características en un plan de implementación aproximado.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Lee este ticket, decide si es un error o una solicitud de característica, "
        f"y enrútalo al subagente correspondiente:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("El botón de inicio de sesión no hace nada en Safari 18."))

La decisión de enrutamiento, qué subagente maneja este ticket, es tomada por el modelo en tiempo de ejecución, no fijada en el código de llamada.
Ambos destinos están disponibles en cada llamada; solo uno (a veces más) se invoca realmente, basándose en el contenido del ticket.
El enrutamiento se adapta a entradas variables; una cadena fija obligaría a cada ticket a pasar por los mismos pasos independientemente del tipo.

5. Despacho de subagentes independientes en paralelo

Ejecuta varias subtareas de forma similar concurrentemente en lugar de una tras otra.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def audit_packages(packages: list[str]) -> None:
    subagents = [
        SubagentConfig(
            name=f"audit-{pkg}",
            description=f"Comprueba el paquete {pkg} en busca de dependencias obsoletas.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Ejecuta todos los subagentes de auditoría para {packages} y lista los hallazgos por paquete."
 
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            print(f"[{message['subagent_name']}] {message['result']}")
 
asyncio.run(audit_packages(["billing", "auth", "search"]))

Un subagente por paquete significa que cada auditoría se ejecuta en su propio contexto, sin estado compartido entre ellos.
Dado que los paquetes no están relacionados, el bucle subyacente puede despachar estas invocaciones de subagente concurrentemente en lugar de serialmente.
Este es el patrón al que recurrir siempre que una tarea sea "haz lo mismo N veces, sobre N entradas independientes".

Relacionado: Creación de subagentes para investigación y delegación paralela - este patrón en profundidad

6. Limitar la profundidad de delegación

Evita que un subagente genere sus propios subagentes más allá de un nivel.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Investiga un tema; no puede delegar más.",
    allowed_tools=["web_search"],
    # No hay campo `subagents` en el propio worker: no tiene capacidad de delegación
    # propia, por lo que la profundidad se limita a un nivel aquí.
)
 
options = AgentOptions(subagents=[worker])

El orquestador puede delegar a research-worker, pero research-worker no tiene subagentes propios a los que delegar.
Limitar la profundidad de esta manera es una elección de diseño deliberada, no un valor predeterminado del SDK; nada te impide anidar subagentes más si lo configuras.
La profundidad sin límites hace que los costos y los modos de fallo sean mucho más difíciles de razonar a medida que un sistema crece.

Relacionado: Barreras de seguridad para sistemas multiagente: Limitación de costos y ámbito - profundidad, acceso a herramientas y límites de gasto combinados

7. Reintentar una llamada fallida a un subagente

Envuelve la invocación de un subagente con un reintento básico en lugar de fallar toda la tarea en un intento fallido.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def run_with_retry(prompt: str, options: AgentOptions, attempts: int = 3):
    last_error = None
    for attempt in range(attempts):
        try:
            results = []
            async for message in query(prompt=prompt, options=options):
                if message.get("type") == "subagent_result":
                    results.append(message["result"])
            return results
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(2 ** attempt)  # retroceso antes de reintentar
    raise RuntimeError(f"Falló después de {attempts} intentos") from last_error

Una llamada fallida a un subagente se trata como cualquier otro fallo que se pueda reintentar, no como un caso especial que el orquestador ignora.
El retroceso exponencial entre intentos evita sobrecargar una dependencia inestable inmediatamente después de que falle.
Los sistemas reales también necesitan un límite en el número total de intentos y una solución alternativa para cuando se agotan los reintentos, cubierto en el artículo de recuperación de errores.

Relacionado: Estrategias de recuperación de errores y reintentos en bucles de agentes - reintentos, soluciones alternativas y disyuntores

Ejemplos intermedios

8. Enrutamiento más despacho en abanico paralelo

Combina una decisión de enrutamiento con un despacho paralelo una vez que se elige el camino.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Cambio pequeño: enruta directamente a un revisor, no se necesita abanico.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Revisa el cambio de un solo archivo.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Usa el subagente reviewer en {changed_files[0]}."
    else:
        # Cambio mayor: abanico de un revisor por archivo, en paralelo.
        options = AgentOptions(
            subagents=[
                SubagentConfig(
                    name=f"reviewer-{i}",
                    description=f"Revisa {f}.",
                    allowed_tools=["file_edit"],
                )
                for i, f in enumerate(changed_files)
            ]
        )
        prompt = f"Usa todos los subagentes de revisión para revisar {changed_files} en paralelo."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

El paso de enrutamiento decide la forma del trabajo (un revisor o muchos) antes de que se ejecute cualquier subagente.
Una vez enrutado al camino de abanico, las subtareas son independientes y se despachan de la misma manera que el ejemplo de auditoría paralela.
La mezcla de enrutamiento y abanico es común en la práctica: el enrutamiento elige la estrategia, el abanico la ejecuta.

9. Un disyuntor alrededor de una herramienta inestable

Deja de reintentar un subagente cuya herramienta subyacente ha fallado repetidamente, en lugar de reintentar indefinidamente.

import asyncio
import time
from claude_agent_sdk import query, AgentOptions
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_after: float = 60.0):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_after = reset_after
        self.opened_at: float | None = None
 
    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.monotonic() - self.opened_at > self.reset_after:
            self.opened_at = None  # semiabierto: permite un intento
            self.failures = 0
            return False
        return True
 
    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()
 
    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None
 
breaker = CircuitBreaker()
 
async def call_flaky_subagent(prompt: str, options: AgentOptions):
    if breaker.is_open():
        raise RuntimeError("Circuito abierto: omitiendo llamada, la dependencia parecía poco saludable")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

El disyuntor rastrea los fallos consecutivos y deja de emitir nuevas llamadas una vez que se cruza un umbral, en lugar de reintentar una dependencia que claramente está caída.
Después de reset_after segundos, permite pasar exactamente un intento (semiabierto) para probar si la dependencia se ha recuperado.
Esto protege al resto del orquestador de gastar tiempo y tokens en llamadas que probablemente volverán a fallar inmediatamente.

10. Puntuación de la salida del orquestador antes de aceptarla

Añade una comprobación ligera que rechaza un resultado combinado obviamente malo en lugar de devolverlo sin verificar.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
def looks_reasonable(result: str) -> bool:
    return len(result.strip()) > 20 and "TODO" not in result
 
async def orchestrate_with_check(prompt: str, options: AgentOptions) -> str:
    final_result = ""
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            final_result = message["result"]
 
    if not looks_reasonable(final_result):
        raise ValueError("El resultado del orquestador falló la comprobación de cordura")
    return final_result

looks_reasonable es un sustituto de cualquier comprobación determinista y ligera que tenga sentido para tu tarea: longitud, campos requeridos, ausencia de texto de marcador de posición.
Comprobar la salida antes de que llegue a un llamador detecta una clase de fallos que un reintento por sí solo no haría, ya que la llamada en sí tuvo éxito, el contenido simplemente estaba mal.
Esta es una vista previa ligera de la disciplina de evaluación cubierta en su totalidad en la lista de verificación de preparación para producción.

Relacionado: Evaluación de la calidad del agente: Una lista de verificación para la preparación de producción - puntuación de la tasa de éxito, el costo y los modos de fallo de forma sistemática

Versiones de la pila: Escrito contra la línea de modelos Claude actual a partir de ~junio de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado), y Claude Haiku 4.5 - y el SDK de Claude Agent (última versión). Los nombres de los modelos, las versiones del SDK y los precios cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.

Conceptos básicos de orquestación agentiva

Buscar en la documentación

Conceptos básicos de orquestación agentiva