Noções Básicas de Orquestração de Agentes

10 exemplos para você começar com orquestração de agentes - 7 básicos e 3 intermediários.

Pré-requisitos

Instale o pacote Python: pip install claude-agent-sdk.
Defina uma chave de API em seu ambiente: export ANTHROPIC_API_KEY=sk-ant-....
Estes exemplos assumem que você já conhece a estrutura de uma única chamada query(); caso contrário, comece primeiro com as noções básicas do Claude Agent SDK.
"Orquestração" aqui significa código de aplicativo (ou um agente de nível superior) que decide quando chamar subagentes e como combinar seus resultados, não qualquer recurso único do próprio SDK.

Exemplos Básicos

1. Um Orquestrador Mínimo: Uma Tarefa, Um Subagente

O menor orquestrador possível: passe uma tarefa para um subagente e leia seu 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="Escreve uma entrada CHANGELOG.md a partir de um resumo de diff.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Use o subagente changelog-writer para adicionar uma entrada CHANGELOG.md para esta versão.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Resultado: {message['result']}")
 
asyncio.run(main())

O orquestrador aqui é uma única chamada query() de nível superior; o próprio modelo decide invocar o único subagente que possui.
Este é todo o padrão: um agente orquestrador, um trabalhador com seu próprio escopo e um resultado que retorna.
Nada disso precisa ser mais complexo até que uma necessidade real de decomposição ou especialização apareça.

Relacionado: Loops de Agente Único vs. Sistemas Multiagentes: Um Modelo Mental - quando este padrão vale a pena ser considerado

2. Mesclando Resultados de Dois Subagentes

Dispare dois subagentes e combine seus resultados em uma única saída.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Resume as alterações de frontend em um PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Resume as alterações de backend em um PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Use ambos os subagentes de resumo, depois escreva uma descrição combinada do PR.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

O modelo orquestrador decide chamar ambos os subagentes e é responsável por combinar seus resultados em uma única descrição.
Cada subagente vê apenas sua própria fatia da tarefa, não a saída do outro, a menos que o orquestrador a passe explicitamente.
Esta é a semente de todos os padrões de fan-out/merge abordados em profundidade mais adiante nesta seção.

3. Uma Cadeia de Prompts Fixa

Execute uma sequência fixa de etapas, cada uma dependendo da 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"Resuma em três marcadores:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Traduza isto para {language}, mantenha como três marcadores:\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("...", "Espanhol")))

Cada etapa é uma chamada query() separada cujo prompt é construído a partir da saída da etapa anterior.
Não há decisão dinâmica aqui sobre qual etapa executar em seguida; a sequência é fixa pelo código de chamada.
Esta é uma cadeia de prompts, o padrão de orquestração mais simples, e muitas vezes tudo o que uma tarefa precisa quando suas etapas são sempre as mesmas e sempre na mesma ordem.

Relacionado: Cadeia de Prompts vs. Roteamento: Escolhendo Seu Padrão de Orquestração - quando uma cadeia fixa é a chamada correta

4. Um Roteador Simples Entre Dois Caminhos

Escolha qual subagente invocar com base na tarefa de entrada, em vez de sempre executar a mesma sequência.

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="Triagem de relatórios de bugs: reproduz, rotula a gravidade.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Escopo de solicitações de recursos em um plano de implementação aproximado.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Leia este ticket, decida se é um bug ou uma solicitação de recurso "
        f"e encaminhe-o para o subagente correspondente:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("O botão de login não faz nada no Safari 18."))

A decisão de roteamento, qual subagente lida com este ticket, é feita pelo modelo em tempo de execução, não fixa no código de chamada.
Ambos os destinos estão disponíveis em todas as chamadas; apenas um (às vezes mais) é realmente invocado, com base no conteúdo do ticket.
O roteamento é adequado para entradas variáveis; uma cadeia fixa forçaria todos os tickets a passar pelas mesmas etapas, independentemente do tipo.

5. Despachando Subagentes Independentes em Paralelo

Execute várias subtarefas de formato semelhante concorrentemente em vez de uma após a outra.

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"Verifica o pacote {pkg} em busca de dependências desatualizadas.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Execute todos os subagentes de auditoria para {packages} e liste as descobertas por pacote."
 
    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"]))

Um subagente por pacote significa que cada auditoria é executada em seu próprio contexto, sem estado compartilhado entre eles.
Como os pacotes não estão relacionados, o loop subjacente pode despachar essas invocações de subagente concorrentemente em vez de serialmente.
Este é o padrão a ser procurado sempre que uma tarefa for "faça a mesma coisa N vezes, em N entradas independentes".

Relacionado: Construindo Subagentes para Pesquisa e Delegação Paralelas - este padrão em profundidade

6. Limitando a Profundidade de Delegação

Impeça que um subagente crie seus próprios subagentes além de um nível.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Pesquisa um tópico; não pode delegar mais.",
    allowed_tools=["web_search"],
    # Nenhum campo `subagents` no próprio worker: ele não tem capacidade de delegação
    # própria, então a profundidade é limitada a um nível aqui.
)
 
options = AgentOptions(subagents=[worker])

O orquestrador pode delegar para research-worker, mas research-worker não tem subagentes próprios para delegar.
Limitar a profundidade dessa forma é uma escolha de design deliberada, não um padrão do SDK; nada impede você de aninhar subagentes mais profundamente se configurar.
Profundidade ilimitada torna os custos e os modos de falha muito mais difíceis de raciocinar à medida que um sistema cresce.

Relacionado: Guardrails para Sistemas Multiagentes: Limitando Custo e Escopo - profundidade, acesso a ferramentas e limites de gastos juntos

7. Tentando Novamente uma Chamada de Subagente Falha

Envolva uma invocação de subagente com uma tentativa básica em vez de falhar toda a tarefa em uma tentativa ruim.

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)  # backoff antes de tentar novamente
    raise RuntimeError(f"Falhou após {attempts} tentativas") from last_error

Uma chamada de subagente com falha é tratada como qualquer outra falha que pode ser retentada, não um caso especial que o orquestrador ignora.
Backoff exponencial entre as tentativas evita sobrecarregar uma dependência instável imediatamente após falhar.
Sistemas reais também precisam de um limite no número total de tentativas e um fallback para quando as tentativas se esgotarem, abordado no artigo de recuperação de erros.

Relacionado: Estratégias de Recuperação de Erros e Retentativas em Loops de Agentes - retentativas, fallbacks e circuit breakers

Exemplos Intermediários

8. Roteamento Mais Fan-Out Paralelo

Combine uma decisão de roteamento com um despacho paralelo assim que o caminho for escolhido.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Mudança pequena: encaminhe diretamente para um revisor, sem necessidade de fan-out.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Revisa a alteração de um único arquivo.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Use o subagente revisor em {changed_files[0]}."
    else:
        # Mudança maior: fan-out um revisor por arquivo, em 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"Use todos os subagentes revisores para revisar {changed_files} em paralelo."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

A etapa de roteamento decide a forma do trabalho (um revisor vs. muitos) antes que qualquer subagente seja executado.
Uma vez roteado para o caminho de fan-out, as subtarefas são independentes e despachadas da mesma forma que o exemplo de auditoria paralela.
Misturar roteamento e fan-out é comum na prática: o roteamento escolhe a estratégia, o fan-out a executa.

9. Um Circuit Breaker em Torno de uma Ferramenta Instável

Pare de tentar um subagente cuja ferramenta subjacente falhou repetidamente, em vez de tentar para sempre.

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  # half-open: permite uma tentativa
            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 aberto: pulando chamada, dependência parecia não saudável")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

O breaker rastreia falhas consecutivas e para de emitir novas chamadas assim que um limite é cruzado, em vez de tentar novamente uma dependência que está claramente inativa.
Após reset_after segundos, ele permite exatamente uma tentativa (half-open) para testar se a dependência se recuperou.
Isso protege o restante do orquestrador de gastar tempo e tokens em chamadas que provavelmente falharão novamente imediatamente.

10. Pontuando a Saída do Orquestrador Antes de Aceitá-la

Adicione uma verificação leve que rejeita um resultado mesclado obviamente ruim em vez de retorná-lo sem verificação.

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("Resultado do orquestrador falhou na verificação de sanidade")
    return final_result

looks_reasonable é um substituto para qualquer verificação barata e determinística que faça sentido para sua tarefa: comprimento, campos obrigatórios, ausência de texto de espaço reservado.
Verificar a saída antes que ela chegue a um chamador captura uma classe de falhas que uma retentativa sozinha não resolveria, pois a chamada em si foi bem-sucedida, o conteúdo estava apenas incorreto.
Esta é uma prévia leve da disciplina de avaliação abordada integralmente na lista de verificação de prontidão para produção.

Relacionado: Avaliando a Qualidade do Agente: Uma Lista de Verificação para Prontidão de Produção - pontuando taxa de sucesso, custo e modos de falha sistematicamente

Versões da Stack: Escrito com base na linha de modelos Claude atual em aproximadamente junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão) e Claude Haiku 4.5 - e o Claude Agent SDK (última versão). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.

Noções Básicas de Orquestração de Agentes

10 exemplos para você começar com orquestração de agentes - 7 básicos e 3 intermediários.

Pré-requisitos

Instale o pacote Python: pip install claude-agent-sdk.
Defina uma chave de API em seu ambiente: export ANTHROPIC_API_KEY=sk-ant-....
Estes exemplos assumem que você já conhece a estrutura de uma única chamada query(); caso contrário, comece primeiro com as noções básicas do Claude Agent SDK.
"Orquestração" aqui significa código de aplicativo (ou um agente de nível superior) que decide quando chamar subagentes e como combinar seus resultados, não qualquer recurso único do próprio SDK.

Exemplos Básicos

1. Um Orquestrador Mínimo: Uma Tarefa, Um Subagente

O menor orquestrador possível: passe uma tarefa para um subagente e leia seu 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="Escreve uma entrada CHANGELOG.md a partir de um resumo de diff.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Use o subagente changelog-writer para adicionar uma entrada CHANGELOG.md para esta versão.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Resultado: {message['result']}")
 
asyncio.run(main())

O orquestrador aqui é uma única chamada query() de nível superior; o próprio modelo decide invocar o único subagente que possui.
Este é todo o padrão: um agente orquestrador, um trabalhador com seu próprio escopo e um resultado que retorna.
Nada disso precisa ser mais complexo até que uma necessidade real de decomposição ou especialização apareça.

Relacionado: Loops de Agente Único vs. Sistemas Multiagentes: Um Modelo Mental - quando este padrão vale a pena ser considerado

2. Mesclando Resultados de Dois Subagentes

Dispare dois subagentes e combine seus resultados em uma única saída.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Resume as alterações de frontend em um PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Resume as alterações de backend em um PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Use ambos os subagentes de resumo, depois escreva uma descrição combinada do PR.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

O modelo orquestrador decide chamar ambos os subagentes e é responsável por combinar seus resultados em uma única descrição.
Cada subagente vê apenas sua própria fatia da tarefa, não a saída do outro, a menos que o orquestrador a passe explicitamente.
Esta é a semente de todos os padrões de fan-out/merge abordados em profundidade mais adiante nesta seção.

3. Uma Cadeia de Prompts Fixa

Execute uma sequência fixa de etapas, cada uma dependendo da 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"Resuma em três marcadores:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Traduza isto para {language}, mantenha como três marcadores:\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("...", "Espanhol")))

Cada etapa é uma chamada query() separada cujo prompt é construído a partir da saída da etapa anterior.
Não há decisão dinâmica aqui sobre qual etapa executar em seguida; a sequência é fixa pelo código de chamada.
Esta é uma cadeia de prompts, o padrão de orquestração mais simples, e muitas vezes tudo o que uma tarefa precisa quando suas etapas são sempre as mesmas e sempre na mesma ordem.

Relacionado: Cadeia de Prompts vs. Roteamento: Escolhendo Seu Padrão de Orquestração - quando uma cadeia fixa é a chamada correta

4. Um Roteador Simples Entre Dois Caminhos

Escolha qual subagente invocar com base na tarefa de entrada, em vez de sempre executar a mesma sequência.

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="Triagem de relatórios de bugs: reproduz, rotula a gravidade.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Escopo de solicitações de recursos em um plano de implementação aproximado.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Leia este ticket, decida se é um bug ou uma solicitação de recurso "
        f"e encaminhe-o para o subagente correspondente:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("O botão de login não faz nada no Safari 18."))

A decisão de roteamento, qual subagente lida com este ticket, é feita pelo modelo em tempo de execução, não fixa no código de chamada.
Ambos os destinos estão disponíveis em todas as chamadas; apenas um (às vezes mais) é realmente invocado, com base no conteúdo do ticket.
O roteamento é adequado para entradas variáveis; uma cadeia fixa forçaria todos os tickets a passar pelas mesmas etapas, independentemente do tipo.

5. Despachando Subagentes Independentes em Paralelo

Execute várias subtarefas de formato semelhante concorrentemente em vez de uma após a outra.

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"Verifica o pacote {pkg} em busca de dependências desatualizadas.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Execute todos os subagentes de auditoria para {packages} e liste as descobertas por pacote."
 
    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"]))

Um subagente por pacote significa que cada auditoria é executada em seu próprio contexto, sem estado compartilhado entre eles.
Como os pacotes não estão relacionados, o loop subjacente pode despachar essas invocações de subagente concorrentemente em vez de serialmente.
Este é o padrão a ser procurado sempre que uma tarefa for "faça a mesma coisa N vezes, em N entradas independentes".

Relacionado: Construindo Subagentes para Pesquisa e Delegação Paralelas - este padrão em profundidade

6. Limitando a Profundidade de Delegação

Impeça que um subagente crie seus próprios subagentes além de um nível.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Pesquisa um tópico; não pode delegar mais.",
    allowed_tools=["web_search"],
    # Nenhum campo `subagents` no próprio worker: ele não tem capacidade de delegação
    # própria, então a profundidade é limitada a um nível aqui.
)
 
options = AgentOptions(subagents=[worker])

O orquestrador pode delegar para research-worker, mas research-worker não tem subagentes próprios para delegar.
Limitar a profundidade dessa forma é uma escolha de design deliberada, não um padrão do SDK; nada impede você de aninhar subagentes mais profundamente se configurar.
Profundidade ilimitada torna os custos e os modos de falha muito mais difíceis de raciocinar à medida que um sistema cresce.

Relacionado: Guardrails para Sistemas Multiagentes: Limitando Custo e Escopo - profundidade, acesso a ferramentas e limites de gastos juntos

7. Tentando Novamente uma Chamada de Subagente Falha

Envolva uma invocação de subagente com uma tentativa básica em vez de falhar toda a tarefa em uma tentativa ruim.

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)  # backoff antes de tentar novamente
    raise RuntimeError(f"Falhou após {attempts} tentativas") from last_error

Uma chamada de subagente com falha é tratada como qualquer outra falha que pode ser retentada, não um caso especial que o orquestrador ignora.
Backoff exponencial entre as tentativas evita sobrecarregar uma dependência instável imediatamente após falhar.
Sistemas reais também precisam de um limite no número total de tentativas e um fallback para quando as tentativas se esgotarem, abordado no artigo de recuperação de erros.

Relacionado: Estratégias de Recuperação de Erros e Retentativas em Loops de Agentes - retentativas, fallbacks e circuit breakers

Exemplos Intermediários

8. Roteamento Mais Fan-Out Paralelo

Combine uma decisão de roteamento com um despacho paralelo assim que o caminho for escolhido.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Mudança pequena: encaminhe diretamente para um revisor, sem necessidade de fan-out.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Revisa a alteração de um único arquivo.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Use o subagente revisor em {changed_files[0]}."
    else:
        # Mudança maior: fan-out um revisor por arquivo, em 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"Use todos os subagentes revisores para revisar {changed_files} em paralelo."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

A etapa de roteamento decide a forma do trabalho (um revisor vs. muitos) antes que qualquer subagente seja executado.
Uma vez roteado para o caminho de fan-out, as subtarefas são independentes e despachadas da mesma forma que o exemplo de auditoria paralela.
Misturar roteamento e fan-out é comum na prática: o roteamento escolhe a estratégia, o fan-out a executa.

9. Um Circuit Breaker em Torno de uma Ferramenta Instável

Pare de tentar um subagente cuja ferramenta subjacente falhou repetidamente, em vez de tentar para sempre.

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  # half-open: permite uma tentativa
            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 aberto: pulando chamada, dependência parecia não saudável")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

O breaker rastreia falhas consecutivas e para de emitir novas chamadas assim que um limite é cruzado, em vez de tentar novamente uma dependência que está claramente inativa.
Após reset_after segundos, ele permite exatamente uma tentativa (half-open) para testar se a dependência se recuperou.
Isso protege o restante do orquestrador de gastar tempo e tokens em chamadas que provavelmente falharão novamente imediatamente.

10. Pontuando a Saída do Orquestrador Antes de Aceitá-la

Adicione uma verificação leve que rejeita um resultado mesclado obviamente ruim em vez de retorná-lo sem verificação.

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("Resultado do orquestrador falhou na verificação de sanidade")
    return final_result

looks_reasonable é um substituto para qualquer verificação barata e determinística que faça sentido para sua tarefa: comprimento, campos obrigatórios, ausência de texto de espaço reservado.
Verificar a saída antes que ela chegue a um chamador captura uma classe de falhas que uma retentativa sozinha não resolveria, pois a chamada em si foi bem-sucedida, o conteúdo estava apenas incorreto.
Esta é uma prévia leve da disciplina de avaliação abordada integralmente na lista de verificação de prontidão para produção.

Relacionado: Avaliando a Qualidade do Agente: Uma Lista de Verificação para Prontidão de Produção - pontuando taxa de sucesso, custo e modos de falha sistematicamente

Versões da Stack: Escrito com base na linha de modelos Claude atual em aproximadamente junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão) e Claude Haiku 4.5 - e o Claude Agent SDK (última versão). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.

Noções Básicas de Orquestração de Agentes

Buscar na documentação

Noções Básicas de Orquestração de Agentes