Busque em todas as páginas da documentação
Quando um orquestrador passa trabalho para um subagente ou uma ferramenta, essa chamada pode falhar completamente, travar para sempre ou retornar com um resultado que parece bom, mas está incorreto ou incompleto. Esta página cobre como detectar cada um desses modos de falha dentro de um loop de agente e responder com retentativas, fallbacks e circuit breakers, para que uma chamada ruim não derrube toda a execução.
Um loop de agente que delega para subagentes ou ferramentas tem uma nova superfície de falha além de "a chamada da API gerou um erro": um subagente pode travar e nunca retornar, ou retornar um resultado que é tecnicamente bem-sucedido, mas inútil.
Retentar cegamente não é uma estratégia. Uma retentativa só ajuda em falhas transitórias, piora uma dependência quebrada ao bombardeá-la e pode duplicar efeitos colaterais se a chamada subjacente não foi idempotente.
Esta página trata a recuperação de erros em nível de orquestração como três decisões em camadas: retentar a mesma chamada um número limitado de vezes com backoff, recorrer a um caminho mais barato ou diferente quando as retentativas se esgotarem e parar de chamar uma dependência completamente assim que ela falhar claramente além de um limite, revisitando-a mais tarde.
Ela também cobre a detecção de um subagente travado, aquele que nunca gera um erro e nunca retorna, pois esse modo de falha requer um timeout explícito em vez de um manipulador de exceções.
Isso está escopado para falhas dentro da própria fronteira de orquestração: uma invocação de subagente, uma chamada de ferramenta, um loop filho travado. Não se trata da mecânica de retentar uma chamada bruta da API Claude, que Estratégias de Retentativa e Backoff Exponencial para Falhas Transitórias da API Claude cobre separadamente.
Cartão de receita de referência rápida - pronto para copiar e colar.
import asyncio
import random
from claude_agent_sdk import query, AgentOptions, SubagentConfig
async def call_subagent_with_retry(prompt, options, *, max_attempts=3, timeout=45.0):
for attempt in range(1, max_attempts + 1):
try:
return await asyncio.wait_for(
run_subagent(prompt, options), timeout=timeout
)
except (asyncio.TimeoutError, RuntimeError):
if attempt == max_attempts:
raise
delay = (2 ** (attempt - 1)) + random.uniform(0, 0.5)
await asyncio.sleep(delay)
async def run_subagent(prompt, options):
result = None
async for message in query(prompt=prompt, options=options):
if message.get("type") == "subagent_result":
result = message
if result is None:
raise RuntimeError("subagent produced no result")
return resultQuando usar isso:
import asyncio
import random
import time
from dataclasses import dataclass, field
from claude_agent_sdk import query, AgentOptions, SubagentConfig
class CircuitOpenError(Exception):
"""Levantada quando um circuit breaker recusa uma chamada porque sua dependência está inativa."""
@dataclass
class CircuitBreaker:
"""Circuit breaker por dependência com uma sonda semiaberta baseada em cooldown."""
failure_threshold: int = 3
cooldown_seconds: float = 30.0
_failures: int = field(default=0, init=False)
_opened_at: float | None = field(default=None, init=False)
def before_call(self) -> None:
if self._opened_at is None:
return
elapsed = time.monotonic() - self._opened_at
if elapsed < self.cooldown_seconds:
raise CircuitOpenError(
f"circuito aberto, tente novamente em {self.cooldown_seconds - elapsed:.1f}s"
)
# Cooldown decorrido: permita que exatamente uma sonda semiaberta passe abaixo.
def record_success(self) -> None:
self._failures = 0
self._opened_at = None
def record_failure(self) -> None:
self._failures += 1
if self._failures >= self.failure_threshold and self._opened_at is None:
self._opened_at = time.monotonic()
async def run_subagent_with_timeout(prompt: str, options: AgentOptions, *, per_message_timeout: float = 20.0):
"""Consome mensagens de query(), levantando TimeoutError se o stream travar entre mensagens."""
stream = query(prompt=prompt, options=options).__aiter__()
result = None
while True:
try:
message = await asyncio.wait_for(stream.__anext__(), timeout=per_message_timeout)
except StopAsyncIteration:
break
if message.get("type") == "subagent_result":
result = message
if result is None:
raise RuntimeError("subagente produziu nenhum resultado utilizável")
return result
async def call_with_retry(make_call, *, max_attempts: int = 3, base_delay: float = 1.0):
"""Tenta novamente uma chamada async com backoff exponencial com jitter."""
last_error: Exception | None = None
for attempt in range(1, max_attempts + 1):
try:
return await make_call()
except (asyncio.TimeoutError, RuntimeError) as exc:
last_error = exc
if attempt == max_attempts:
break
delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
await asyncio.sleep(delay)
raise last_error
# Um circuit breaker por dependência, chaveado pelo nome do subagente.
breakers = {
"deep-research": CircuitBreaker(failure_threshold=3, cooldown_seconds=30.0),
"quick-search": CircuitBreaker(failure_threshold=3, cooldown_seconds=15.0),
}
DEEP_RESEARCH = SubagentConfig(
name="deep-research",
description="Executa uma passagem de pesquisa completa e multi-etapas e cita fontes.",
allowed_tools=["web_search", "web_fetch"],
)
QUICK_SEARCH = SubagentConfig(
name="quick-search",
description="Executa uma única pesquisa web rápida e resume os principais resultados.",
allowed_tools=["web_search"],
)
async def research_topic(topic: str) -> dict:
"""Tenta o subagente completo primeiro; recorre ao subagente barato em caso de falha repetida."""
for name, subagent, timeout in [
("deep-research", DEEP_RESEARCH, 60.0),
("quick-search", QUICK_SEARCH, 20.0),
]:
breaker = breakers[name]
try:
breaker.before_call()
except CircuitOpenError:
print(f"[{name}] circuito aberto, pulando direto para o fallback")
continue
options = AgentOptions(allowed_tools=[], subagents=[subagent])
try:
result = await call_with_retry(
lambda: run_subagent_with_timeout(
f"Use o subagente {name} para pesquisar: {topic}",
options,
per_message_timeout=timeout,
),
max_attempts=2,
)
breaker.record_success()
result["served_by"] = name
return result
except (asyncio.TimeoutError, RuntimeError, CircuitOpenError) as exc:
breaker.record_failure()
print(f"[{name}] falhou após retentativas: {exc}, tentando o próximo caminho")
continue
raise RuntimeError(f"todos os caminhos de pesquisa falharam para o tópico: {topic}")
asyncio.run(research_topic("recent developments in battery recycling"))O que isso demonstra:
query() para capturar um subagente que trava sem gerar um erro.call_with_retry aplicando backoff exponencial com jitter a qualquer chamável, mantido separado da detecção de travamento específica do subagente.CircuitBreaker com chave por nome de subagente, para que a falha de deep-research não afete o circuit breaker independente de quick-search.served_by, para que os chamadores possam distinguir uma resposta degradada da resposta primária.query() pode falhar de três maneiras distintas: ela gera uma exceção (um erro de ferramenta, um resultado malformado), ela trava (nenhuma mensagem chega e nenhuma exceção é gerada), ou ela é concluída com um resultado que é tecnicamente válido, mas vazio ou incorreto.asyncio.wait_for é o que transforma um travamento em algo que a lógica de retentativa pode capturar.before_call() encurta o caminho imediatamente assim que o limite de falha é cruzado, então um subagente comprovadamente ruim para de absorver orçamento de retentativa.before_call() passa, e o resultado dessa única sonda (record_success ou record_failure) decide se o circuito fecha novamente ou reinicia seu cooldown.| Modo de Falha | Sintoma | Resposta Correta |
|---|---|---|
| Erro transitório de ferramenta | Exceção gerada, provavelmente terá sucesso em uma segunda tentativa | Retentar com backoff |
| Subagente travado | Nenhuma mensagem, nenhuma exceção, a chamada simplesmente trava | Timeout via asyncio.wait_for, então trate como uma falha |
| Resultado ruim/incompleto | Chamada retorna normalmente, mas o resultado está vazio ou incorreto | Valide o resultado explicitamente; trate a falha de validação como uma falha de chamada |
| Dependência inativa por um tempo | Várias falhas consecutivas contra o mesmo subagente/ferramenta | Circuit breaker, pare de chamá-lo até que o cooldown passe |
| Caminho primário esgotado | Retentativas e circuit breaker dizem não para esta chamada | Recorrer a um subagente mais simples ou diferente |
# asyncio.wait_for só ajuda se você aplicá-lo a algo que pode realmente
# ser interrompido entre awaits. Envolver todo o loop `async for` só
# faz timeout do loop como um todo, não de cada mensagem individualmente travada, então
# chame __anext__() diretamente por mensagem quando precisar de granularidade por mensagem.
stream = query(prompt=prompt, options=options).__aiter__()
message = await asyncio.wait_for(stream.__anext__(), timeout=20.0)
# StopAsyncIteration significa que o stream terminou normalmente, não que falhou;
# não deixe cair na mesma cláusula except que TimeoutError/RuntimeError.| Parâmetro | Tipo | Descrição |
|---|---|---|
failure_threshold | int | Falhas consecutivas antes que o CircuitBreaker abra |
cooldown_seconds | float | Quanto tempo o circuito permanece aberto antes de permitir uma sonda semiaberta |
max_attempts | int | Tentativas totais (incluindo a primeira) para call_with_retry |
base_delay | float | Segundos base para backoff exponencial, antes do jitter |
per_message_timeout | float | Segundos para esperar pela próxima mensagem antes de tratar o subagente como travado |
per_message_timeout mais curto que a latência normal do subagente mata chamadas legítimas, porém lentas, que então se parecem com um travamento real. Correção: defina o timeout a partir do perfil de latência real da tarefa, não um palpite, e dê mais margem para um subagente do tipo deep-research do que para um do tipo quick-search.call_with_retry faz acima.served_by acima) e registre ou alerte sobre a taxa de fallback, não apenas engula a falha.record_success() nunca for chamado na sonda que passa após o cooldown, o breaker nunca poderá realmente fechar novamente, mesmo quando a dependência se recuperar. Correção: sempre roteie ambos os resultados da chamada da sonda através do mesmo caminho record_success/record_failure como uma chamada normal.| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
| Loop de retentativa manual em cada local de chamada | Um único script descartável com uma chamada para proteger | O padrão precisa ser reutilizado em muitos locais de chamada de subagente/ferramenta |
Uma biblioteca de retentativa/circuit breaker (ex: tenacity) | Você quer primitivas de backoff e breaker testadas em batalha em vez de manter as suas | Você precisa de semânticas de timeout específicas para um gerador assíncrono travado, o que a maioria dessas bibliotecas não modela diretamente |
| Deixar o loop interno do subagente retentar suas chamadas de ferramenta | A falha está dentro do uso de ferramenta do próprio subagente, não na fronteira de delegação pai-filho | O próprio subagente travou ou falhou em retornar; apenas o pai pode observar e recuperar isso |
| Falhar rapidamente para um ponto de verificação humano | A ação é de alto risco ou tem efeitos colaterais onde uma retentativa cega ou fallback é insegura | A falha é um erro transitório rotineiro onde uma retentativa automatizada é segura e barata |
try/except.Não há um número universal, mas 2-3 tentativas totais é um padrão razoável para retentativas em nível de orquestração. Mais do que isso geralmente significa que a falha não é transitória, e um fallback ou circuit breaker é a melhor resposta do que mais retentativas.
Não. Retente apenas chamadas que são seguras para repetir, o que significa chamadas idempotentes ou aquelas com uma chave de deduplicação. Retentar uma chamada com um efeito colateral não dedupicado arrisca executar esse efeito colateral mais de uma vez.
Uma retentativa repete a mesma chamada para a mesma dependência, esperando que a falha tenha sido transitória. Um fallback muda para um caminho completamente diferente, um subagente mais simples, uma ferramenta mais barata, depois que você decidiu que o caminho primário não vai ter sucesso desta vez.
Geralmente sim, por dependência. Se dez tarefas concorrentes estão cada uma chamando o mesmo subagente, elas devem compartilhar um breaker para esse subagente para que o circuito reflita a saúde real da dependência, não a visão privada de cada tarefa sobre ela.
Assim que o cooldown decorre, o breaker permite que exatamente uma chamada passe como uma sonda. Se essa sonda for bem-sucedida, o breaker fecha e retoma as chamadas normais. Se falhar, o breaker reabre e o cooldown recomeça.
Envolva o consumo do stream de mensagens em asyncio.wait_for, aplicado por mensagem em vez de ao redor de todo o loop, para que uma lacuna maior que seu timeout entre mensagens gere TimeoutError que você possa capturar e tratar como uma falha.
Na fronteira que realmente observou a falha. Falhas de delegação (uma chamada de subagente que travou ou gerou erro) pertencem à lógica de retentativa do orquestrador pai. Falhas de ferramenta do próprio subagente pertencem ao loop interno desse subagente, não duplicadas pelo pai.
Sim, para qualquer coisa executada com concorrência. Sem jitter, muitos chamadores retentando a mesma dependência após uma falha compartilhada tendem a retentar em sincronia, produzindo um novo pico de carga em cada intervalo de backoff em vez de espalhar as retentativas.
Algo significativamente mais barato, mais simples ou com escopo diferente do caminho primário, não apenas uma cópia da mesma chamada. Um fallback que falha da mesma forma que o primário adiciona latência sem adicionar resiliência.
Registre uma falha do breaker por sequência de tentativas de chamada que falhou ultimamente, não uma por retentativa individual dentro dessa sequência. O exemplo acima chama record_failure() uma vez, depois que call_with_retry esgotou suas tentativas, não dentro do próprio loop de retentativa.
Sim, e é necessário. Um subagente pode retornar normalmente com um resultado vazio ou incorreto, que não gerará um erro por si só. Validar explicitamente o resultado e gerar um erro se a validação falhar é o que permite que a lógica de retentativa, fallback e circuit breaker o trate da mesma forma que qualquer outra falha.
Versões da Stack: Escrito contra a linha de modelos Claude atual em ~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.
Revisado por Chris St. John·Última atualização: 16 de jul. de 2026