# Como implementar circuit breaker em integrações com API de CPF

> Aprenda a implementar o padrão circuit breaker em integrações com API de CPF. Proteja sua aplicação contra falhas em cascata.

**Publicado:** 14/08/2026
**Autor:** Redação CPFHub.io
**URL:** https://cpfhub.io/blog/como-implementar-circuit-breaker-integracoes-api-cpf

---


O padrão circuit breaker protege sua aplicação contra falhas em cascata ao monitorar erros consecutivos na chamada à API de CPF e bloquear automaticamente novas requisições quando o threshold é atingido — retornando um fallback imediato em vez de deixar o sistema acumular timeouts. Para integrações com a API da CPFHub.io, que tem latência média de ~900ms, um circuit breaker com threshold de 5 falhas e recovery timeout de 60 segundos é um bom ponto de partida. O artigo de Martin Fowler sobre [CircuitBreaker](https://martinfowler.com/bliki/CircuitBreaker.html) é a referência canônica do padrão.

---

## O que é o padrão circuit breaker

O circuit breaker funciona como um disjuntor elétrico. Quando detecta que uma dependência externa está falhando repetidamente, ele "abre o circuito" e para de enviar requisições, evitando que a falha se propague.

### Os três estados

- **Fechado (Closed):** tudo funciona normalmente. As requisições passam para a API e as respostas são retornadas ao chamador.
- **Aberto (Open):** a API está falhando. O circuit breaker bloqueia as requisições imediatamente, sem nem tentar chamar a API, retornando um erro ou resposta padrão.
- **Meio-aberto (Half-Open):** após um período de espera, o circuit breaker permite que uma requisição de teste passe. Se ela for bem-sucedida, o circuito fecha novamente. Se falhar, volta a abrir.

### Por que isso importa para consulta de CPF

Quando sua aplicação depende de uma API externa para consultar CPFs, qualquer instabilidade nessa dependência pode se propagar e derrubar todo o sistema. Um timeout que deveria afetar apenas a validação de CPF acaba travando filas, acumulando conexões e degradando a experiência do usuário em funcionalidades que nem dependem dessa consulta. O circuit breaker existe para prevenir exatamente esse tipo de falha em cascata — e é especialmente relevante para APIs com latência na casa de ~900ms, onde timeouts acumulados afetam rapidamente a performance geral.

---

## Implementação em Python

Vamos construir um circuit breaker do zero, sem dependências externas, para entender cada componente:

```python
import time
import requests
import logging
from enum import Enum
from threading import Lock

logger = logging.getLogger(__name__)

class CircuitState(Enum):
 CLOSED = "closed"
 OPEN = "open"
 HALF_OPEN = "half_open"

class CircuitBreaker:
 """Circuit breaker para proteger chamadas à API de CPF."""

 def __init__(
 self,
 failure_threshold: int = 5,
 recovery_timeout: int = 60,
 half_open_max_calls: int = 1
 ):
 self.failure_threshold = failure_threshold
 self.recovery_timeout = recovery_timeout
 self.half_open_max_calls = half_open_max_calls

 self._state = CircuitState.CLOSED
 self._failure_count = 0
 self._last_failure_time = 0
 self._half_open_calls = 0
 self._lock = Lock()

 @property
 def state(self) -> CircuitState:
 with self._lock:
 if self._state == CircuitState.OPEN:
 if time.time() - self._last_failure_time >= self.recovery_timeout:
 self._state = CircuitState.HALF_OPEN
 self._half_open_calls = 0
 logger.info("Circuit breaker transicionou para HALF_OPEN")
 return self._state

 def record_success(self):
 with self._lock:
 self._failure_count = 0
 if self._state == CircuitState.HALF_OPEN:
 self._state = CircuitState.CLOSED
 logger.info("Circuit breaker fechou -- API recuperada")

 def record_failure(self):
 with self._lock:
 self._failure_count += 1
 self._last_failure_time = time.time()

 if self._failure_count >= self.failure_threshold:
 self._state = CircuitState.OPEN
 logger.warning(
 f"Circuit breaker ABERTO apos {self._failure_count} falhas consecutivas"
 )

 def allow_request(self) -> bool:
 current_state = self.state
 if current_state == CircuitState.CLOSED:
 return True
 if current_state == CircuitState.HALF_OPEN:
 with self._lock:
 if self._half_open_calls < self.half_open_max_calls:
 self._half_open_calls += 1
 return True
 return False
 return False

class CPFClientWithCircuitBreaker:
 """Cliente de API de CPF com circuit breaker integrado."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 TIMEOUT = 30

 def __init__(self, api_key: str):
 self.api_key = api_key
 self.circuit = CircuitBreaker(
 failure_threshold=5,
 recovery_timeout=60
 )
 self.session = requests.Session()
 self.session.headers.update({
 "x-api-key": self.api_key,
 "Accept": "application/json"
 })

 def consultar(self, cpf: str) -> dict:
 if not self.circuit.allow_request():
 logger.warning("Circuit breaker ABERTO -- requisicao bloqueada")
 return {"success": False, "error": "circuit_open"}

 try:
 response = self.session.get(
 f"{self.BASE_URL}/{cpf}",
 timeout=self.TIMEOUT
 )
 response.raise_for_status()
 resultado = response.json()

 self.circuit.record_success()
 return resultado

 except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
 self.circuit.record_failure()
 logger.error(f"Falha na API de CPF: {e}")
 return {"success": False, "error": str(e)}

 except requests.exceptions.HTTPError as e:
 if response.status_code >= 500:
 self.circuit.record_failure()
 return {"success": False, "error": str(e)}
```

### Uso do cliente

```python
cliente = CPFClientWithCircuitBreaker(api_key="SUA_CHAVE_API")

resultado = cliente.consultar("12345678900")
if resultado.get("success"):
 print(f"Nome: {resultado['data']['name']}")
elif resultado.get("error") == "circuit_open":
 print("Servico temporariamente indisponivel. Tente novamente em breve.")
else:
 print(f"Erro na consulta: {resultado.get('error')}")
```

---

## Implementação em Node.js

Para aplicações JavaScript, a biblioteca `opossum` facilita a implementação:

```javascript
const CircuitBreaker = require("opossum");
const axios = require("axios");

const API_KEY = "SUA_CHAVE_API";
const BASE_URL = "https://api.cpfhub.io/cpf";
const TIMEOUT_MS = 30000;

async function consultarCPF(cpf) {
 const response = await axios.get(`${BASE_URL}/${cpf}`, {
 headers: {
 "x-api-key": API_KEY,
 "Accept": "application/json"
 },
 timeout: TIMEOUT_MS
 });
 return response.data;
}

const circuitOptions = {
 timeout: TIMEOUT_MS,
 errorThresholdPercentage: 50,
 resetTimeout: 60000,
 volumeThreshold: 5
};

const breaker = new CircuitBreaker(consultarCPF, circuitOptions);

breaker.on("open", () => console.warn("Circuit breaker ABERTO"));
breaker.on("halfOpen", () => console.info("Circuit breaker HALF_OPEN"));
breaker.on("close", () => console.info("Circuit breaker FECHADO"));
breaker.fallback(() => ({
 success: false,
 error: "circuit_open"
}));

// Uso
breaker.fire("12345678900")
 .then(result => {
 if (result.success) {
 console.log(`Nome: ${result.data.name}`);
 } else {
 console.log("Fallback ativado:", result.error);
 }
 })
 .catch(err => console.error("Erro:", err.message));
```

---

## Configurações recomendadas

Os valores ideais dependem do contexto da sua aplicação, mas estas são boas referências para integrações com API de CPF:

| Parâmetro | Valor recomendado | Justificativa |
|------------------------|-------------------|-------------------------------------------------|
| Failure threshold | 5 falhas | Evita abrir o circuito por falhas isoladas |
| Recovery timeout | 60 segundos | Tempo suficiente para a maioria dos incidentes |
| Timeout da requisição | 30 segundos | Margem generosa sobre a latência média de ~900ms |
| Half-open max calls | 1-3 | Testa a recuperação sem sobrecarregar |

---

## O que fazer quando o circuito está aberto

Quando o circuit breaker abre, sua aplicação precisa de uma estratégia de fallback. As opções mais comuns são:

### 1. Retornar erro amigável

Informe ao usuário que a validação está temporariamente indisponível e que ele pode prosseguir (se o fluxo permitir).

### 2. Validação local apenas

Use a validação algorítmica do CPF (dígitos verificadores) como fallback. Ela não confirma a identidade, mas garante que o formato é válido.

### 3. Enfileirar para processamento posterior

Salve o CPF em uma fila e processe a consulta quando o circuito fechar. Essa abordagem é ideal para fluxos que não precisam da resposta em tempo real.

---

## Monitorando o circuit breaker

Registre métricas sobre o comportamento do circuit breaker:

- **Número de vezes que o circuito abriu** por período.
- **Tempo médio no estado aberto** antes de recuperar.
- **Taxa de falhas que disparam a abertura** versus falhas isoladas.

Essas métricas ajudam a ajustar os parâmetros e a identificar padrões de instabilidade.

---

## Testando o circuit breaker

Para validar que o circuit breaker funciona corretamente, simule cenários de falha:

```python
import unittest
from unittest.mock import patch, MagicMock

class TestCircuitBreaker(unittest.TestCase):

 def test_abre_apos_threshold(self):
 cliente = CPFClientWithCircuitBreaker(api_key="teste")

 with patch.object(cliente.session, "get", side_effect=requests.exceptions.Timeout):
 for _ in range(5):
 cliente.consultar("12345678900")

 self.assertEqual(cliente.circuit.state, CircuitState.OPEN)

 def test_bloqueia_quando_aberto(self):
 cliente = CPFClientWithCircuitBreaker(api_key="teste")
 cliente.circuit._state = CircuitState.OPEN
 cliente.circuit._last_failure_time = time.time()

 resultado = cliente.consultar("12345678900")
 self.assertEqual(resultado["error"], "circuit_open")
```

---

## Perguntas frequentes

### Quantas falhas consecutivas devo configurar para abrir o circuito?
O threshold recomendado é de 5 falhas consecutivas para integrações com API de CPF. Valores muito baixos (1-2) abrem o circuito por falhas isoladas que a API já resolveria com um simples retry. Valores muito altos (10+) permitem que a falha persista por tempo demais antes de acionar o fallback. Ajuste com base nos dados de monitoramento: se o circuito abre com frequência por falhas isoladas, aumente o threshold.

### O circuit breaker substitui o retry com backoff exponencial?
Não — os dois padrões são complementares. O retry com backoff exponencial lida com falhas isoladas e transitórias, tentando a mesma requisição algumas vezes com intervalo crescente. O circuit breaker atua quando as falhas são sistêmicas e consecutivas, bloqueando o fluxo para proteger o sistema. A combinação ideal é: retry com até 3 tentativas, e circuit breaker que abre após 5 falhas acumuladas.

### Como definir o recovery timeout do circuit breaker?
O recovery timeout (tempo que o circuito permanece aberto antes de testar novamente) deve ser longo o suficiente para a API se recuperar de incidentes típicos, mas curto o suficiente para não deixar o sistema no fallback por muito tempo. Para a API da CPFHub.io, 60 segundos é um bom padrão. Se incidentes costumam durar mais tempo na sua experiência, considere aumentar para 120-300 segundos.

### O circuit breaker precisa ser compartilhado entre threads ou instâncias?
Sim, para funcionar corretamente. Um circuit breaker por instância de processo — usando um singleton com lock de thread (como no exemplo Python acima) — garante que o estado seja consistente dentro do processo. Em arquiteturas com múltiplas instâncias da aplicação, considere um circuit breaker distribuído com estado em Redis para que todas as instâncias compartilhem o mesmo estado de abertura.

### Leia também

- [SLA de API de CPF: entenda os níveis de disponibilidade](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade)
- [Como implementar retry com backoff exponencial em consultas de API de CPF](https://cpfhub.io/blog/como-implementar-retry-backoff-exponencial-consultas-api-cpf)
- [Como fazer stress test em integrações com API de CPF antes de ir para produção](https://cpfhub.io/blog/como-fazer-stress-test-em-integracoes-com-api-de-cpf-antes-de-ir-para-producao)
- [Como implementar cache inteligente para respostas da API de CPF](https://cpfhub.io/blog/como-implementar-cache-inteligente-respostas-api-cpf)

---

## Conclusão

O circuit breaker é uma camada essencial de resiliência para qualquer aplicação que depende de APIs externas. Ao implementá-lo em integrações com a API de CPF, você protege sua aplicação contra falhas em cascata, melhora a experiência do usuário e ganha visibilidade sobre a saúde das suas dependências.

A [**CPFHub.io**](https://www.cpfhub.io/)

Cadastre-se em [cpfhub.io](https://www.cpfhub.io/)