# Como criar um SDK interno para padronizar consultas de CPF na sua empresa

> Aprenda a criar um SDK interno que padroniza consultas de CPF via API em todos os times da sua empresa. Código reutilizável e consistente.

**Publicado:** 23/08/2026
**Autor:** Redação CPFHub.io
**URL:** https://cpfhub.io/blog/como-criar-sdk-interno-padronizar-consultas-cpf-empresa

---


Quando múltiplos times na mesma empresa integram a API de CPF de forma independente, o resultado inevitável é inconsistência. Um time implementa retry, outro não. Um trata erros de forma detalhada, outro engole exceções silenciosamente. Um mascara CPFs nos logs, outro os expõe em texto claro. A solução é criar um SDK interno — uma biblioteca compartilhada que encapsula toda a lógica de integração com a API de CPF em um único lugar, integrando-o com a API da [**CPFHub.io**](https://www.cpfhub.io/).

---

## Por que criar um SDK interno

### Problemas que o SDK resolve

- **Duplicação de código:** cada time escreve sua própria integração, multiplicando pontos de manutenção.
- **Inconsistência no tratamento de erros:** sem padrão, cada integração lida com falhas de forma diferente.
- **Falta de observabilidade:** sem métricas centralizadas, é impossível ter visão global do consumo.
- **Risco de compliance:** se um time não mascara CPFs nos logs, toda a empresa está em risco perante a LGPD.
- **Onboarding lento:** novos desenvolvedores precisam entender a API do zero em vez de usar uma abstração pronta.

### Benefícios diretos

- Uma linha de código para consultar CPF.
- Retry, timeout e circuit breaker embutidos.
- Logs padronizados e conformes com LGPD.
- Métricas prontas para exportação.
- Atualizações centralizadas -- corrigir uma vez, beneficiar todos os times.

---

## Arquitetura do SDK

O SDK deve ter quatro camadas:

1. **Cliente HTTP:** faz a chamada à API com timeout configurado.
2. **Resiliência:** retry com backoff exponencial e circuit breaker.
3. **Observabilidade:** logging estruturado e métricas.
4. **Interface pública:** métodos simples que os times consumirão.

---

## Implementação em Python

### Estrutura do pacote

```
cpfhub-sdk/
 cpfhub_sdk/
 __init__.py
 client.py
 retry.py
 exceptions.py
 setup.py
```

### Exceções personalizadas

```python
# cpfhub_sdk/exceptions.py

class CPFHubError(Exception):
 """Erro base do SDK."""
 pass

class CPFHubAuthError(CPFHubError):
 """Erro de autenticação (401)."""
 pass

class CPFHubQuotaError(CPFHubError):
 """Cota excedida — a API cobra R$0,15 por consulta adicional, não bloqueia."""
 pass

class CPFHubTimeoutError(CPFHubError):
 """Timeout na requisição."""
 pass

class CPFHubServerError(CPFHubError):
 """Erro no servidor (5xx)."""
 pass
```

### Módulo de retry

```python
# cpfhub_sdk/retry.py

import time
import random
from typing import Callable, Any

def com_retry(
 func: Callable,
 max_retries: int = 3,
 base_delay: float = 1.0,
 max_delay: float = 30.0,
 retryable_exceptions: tuple = ()
) -> Any:
 """Executa função com retry e backoff exponencial."""
 for tentativa in range(max_retries + 1):
 try:
 return func()
 except retryable_exceptions as e:
 if tentativa >= max_retries:
 raise
 delay = min(base_delay * (2 ** tentativa) * (0.5 + random.random()), max_delay)
 time.sleep(delay)
```

### Cliente principal

```python
# cpfhub_sdk/client.py

import time
import requests
import logging
import json
from typing import Optional
from dataclasses import dataclass
from .exceptions import (
 CPFHubAuthError,
 CPFHubQuotaError,
 CPFHubTimeoutError,
 CPFHubServerError
)
from .retry import com_retry

logger = logging.getLogger("cpfhub_sdk")

@dataclass
class CPFData:
 """Dados retornados pela API."""
 cpf: str
 name: str
 name_upper: str
 gender: str
 birth_date: str
 day: str
 month: str
 year: str

class CPFHub:
 """SDK para consulta de CPF via CPFHub.io."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 DEFAULT_TIMEOUT = 30
 DEFAULT_MAX_RETRIES = 3

 def __init__(
 self,
 api_key: str,
 timeout: int = DEFAULT_TIMEOUT,
 max_retries: int = DEFAULT_MAX_RETRIES
 ):
 self._api_key = api_key
 self._timeout = timeout
 self._max_retries = max_retries
 self._session = requests.Session()
 self._session.headers.update({
 "x-api-key": self._api_key,
 "Accept": "application/json"
 })

 def consultar(self, cpf: str) -> Optional[CPFData]:
 """
 Consulta um CPF na API da CPFHub.

 Args:
 cpf: Número do CPF (aceita formatado ou apenas dígitos)

 Returns:
 CPFData com os dados ou None se não encontrado

 Raises:
 CPFHubAuthError: se a chave de API for inválida
 CPFHubQuotaError: se a cota mensal foi excedida (a API cobra excedente, não bloqueia)
 CPFHubTimeoutError: se a requisição excedeu o timeout
 CPFHubServerError: se o servidor retornou erro 5xx
 """
 cpf_limpo = self._limpar_cpf(cpf)

 def _fazer_requisicao():
 return self._executar_consulta(cpf_limpo)

 return com_retry(
 _fazer_requisicao,
 max_retries=self._max_retries,
 retryable_exceptions=(CPFHubTimeoutError, CPFHubServerError)
 )

 def _executar_consulta(self, cpf: str) -> Optional[CPFData]:
 """Executa a consulta HTTP."""
 inicio = time.time()

 try:
 response = self._session.get(
 f"{self.BASE_URL}/{cpf}",
 timeout=self._timeout
 )
 except requests.exceptions.Timeout:
 self._log_requisicao(cpf, None, time.time() - inicio, False)
 raise CPFHubTimeoutError(f"Timeout apos {self._timeout}s")
 except requests.exceptions.RequestException as e:
 self._log_requisicao(cpf, None, time.time() - inicio, False)
 raise CPFHubServerError(str(e))

 latencia = time.time() - inicio

 if response.status_code == 401:
 self._log_requisicao(cpf, 401, latencia, False)
 raise CPFHubAuthError("Chave de API invalida")

 if response.status_code >= 500:
 self._log_requisicao(cpf, response.status_code, latencia, False)
 raise CPFHubServerError(f"Erro do servidor: {response.status_code}")

 dados = response.json()
 self._log_requisicao(cpf, response.status_code, latencia, dados.get("success", False))

 if dados.get("success"):
 return CPFData(
 cpf=dados["data"]["cpf"],
 name=dados["data"]["name"],
 name_upper=dados["data"]["nameUpper"],
 gender=dados["data"]["gender"],
 birth_date=dados["data"]["birthDate"],
 day=dados["data"]["day"],
 month=dados["data"]["month"],
 year=dados["data"]["year"]
 )

 return None

 @staticmethod
 def _limpar_cpf(cpf: str) -> str:
 """Remove formatação do CPF."""
 import re
 cpf_limpo = re.sub(r"\D", "", cpf)
 if len(cpf_limpo) != 11:
 raise ValueError(f"CPF deve ter 11 digitos, recebeu {len(cpf_limpo)}")
 return cpf_limpo

 @staticmethod
 def _mascarar_cpf(cpf: str) -> str:
 """Mascara CPF para logs (LGPD)."""
 return f"{cpf[:3]}***{cpf[-2:]}"

 def _log_requisicao(self, cpf, status_code, latencia, sucesso):
 """Registra log estruturado."""
 log_entry = {
 "service": "cpfhub-sdk",
 "cpf": self._mascarar_cpf(cpf),
 "status": status_code,
 "latency_ms": round(latencia * 1000),
 "success": sucesso
 }
 logger.info(json.dumps(log_entry))
```

### Uso pelo time consumidor

```python
from cpfhub_sdk import CPFHub, CPFHubAuthError, CPFHubQuotaError

cliente = CPFHub(api_key="SUA_CHAVE_API")

try:
 dados = cliente.consultar("123.456.789-00")
 if dados:
 print(f"Nome: {dados.name}")
 print(f"Nascimento: {dados.birth_date}")
except CPFHubAuthError:
 print("Chave de API invalida. Verifique as credenciais.")
except CPFHubQuotaError:
 print("Cota excedida. Consultas adicionais serão cobradas a R$0,15 cada.")
```

---

## Implementação em Node.js

```javascript
// cpfhub-sdk/index.js

const axios = require("axios");

class CPFHubError extends Error {
 constructor(message, code) {
 super(message);
 this.name = "CPFHubError";
 this.code = code;
 }
}

class CPFHub {
 constructor({ apiKey, timeout = 30000, maxRetries = 3 }) {
 this.apiKey = apiKey;
 this.timeout = timeout;
 this.maxRetries = maxRetries;
 this.baseUrl = "https://api.cpfhub.io/cpf";
 }

 async consultar(cpf) {
 const cpfLimpo = cpf.replace(/\D/g, "");

 if (cpfLimpo.length !== 11) {
 throw new CPFHubError("CPF deve ter 11 digitos", "INVALID_CPF");
 }

 let lastError;
 for (let tentativa = 0; tentativa <= this.maxRetries; tentativa++) {
 try {
 const response = await axios.get(`${this.baseUrl}/${cpfLimpo}`, {
 headers: {
 "x-api-key": this.apiKey,
 "Accept": "application/json"
 },
 timeout: this.timeout
 });

 if (response.data.success) {
 return response.data.data;
 }
 return null;

 } catch (error) {
 lastError = error;
 const status = error.response ? error.response.status : null;

 if (status === 401) {
 throw new CPFHubError("Chave de API invalida", "AUTH_ERROR");
 }

 if (tentativa < this.maxRetries && (status >= 500 || !status)) {
 const delay = Math.min(1000 * Math.pow(2, tentativa) * (0.5 + Math.random()), 30000);
 await new Promise(r => setTimeout(r, delay));
 }
 }
 }

 throw new CPFHubError(`Falha apos ${this.maxRetries} tentativas`, "MAX_RETRIES");
 }
}

module.exports = { CPFHub, CPFHubError };
```

---

## Versionamento e distribuição

Distribua o SDK como pacote interno:

- **Python:** publique no PyPI privado da empresa ou use instalação direta do repositório Git.
- **Node.js:** publique no npm registry privado ou use `npm link` para desenvolvimento local.

Mantenha versionamento semântico (SemVer) para que os times saibam quando há breaking changes. A [especificação SemVer](https://semver.org/) define como comunicar compatibilidade entre versões de forma clara.

---

## Testes do SDK

```python
import pytest
from unittest.mock import patch, MagicMock
from cpfhub_sdk import CPFHub, CPFHubAuthError

class TestCPFHub:
 def setup_method(self):
 self.cliente = CPFHub(api_key="teste")

 @patch("cpfhub_sdk.client.requests.Session.get")
 def test_consulta_sucesso(self, mock_get):
 mock_response = MagicMock()
 mock_response.status_code = 200
 mock_response.json.return_value = {
 "success": True,
 "data": {
 "cpf": "12345678900", "name": "Teste",
 "nameUpper": "TESTE", "gender": "M",
 "birthDate": "1990-01-01", "day": "01",
 "month": "01", "year": "1990"
 }
 }
 mock_get.return_value = mock_response

 resultado = self.cliente.consultar("12345678900")
 assert resultado.name == "Teste"

 @patch("cpfhub_sdk.client.requests.Session.get")
 def test_auth_error(self, mock_get):
 mock_response = MagicMock()
 mock_response.status_code = 401
 mock_get.return_value = mock_response

 with pytest.raises(CPFHubAuthError):
 self.cliente.consultar("12345678900")
```

---

## Perguntas frequentes

### Por que centralizar a integração com a API de CPF em um SDK interno?

Sem um SDK compartilhado, cada time reimplementa retry, timeout, mascaramento de CPF nos logs e tratamento de erros de forma diferente. Isso multiplica os pontos de falha e aumenta o risco de vazamento de dados pessoais. Um SDK centralizado corrige uma vez e distribui a correção para toda a empresa automaticamente.

### Como o SDK deve tratar o limite de consultas da CPFHub.io?

A API da CPFHub.io não bloqueia requisições ao atingir o limite mensal — ela continua respondendo e cobra R$0,15 por consulta adicional. O SDK pode emitir um alerta de log quando o consumo estiver próximo do limite, mas não precisa implementar bloqueio local. O plano gratuito inclui 50 consultas/mês e o Pro oferece 1.000 por R$149/mês.

### Como garantir conformidade com a LGPD no SDK?

O SDK deve mascarar o CPF em todos os logs antes de gravá-los, retendo apenas os primeiros três e os dois últimos dígitos. Os dados retornados pela API — nome, data de nascimento, gênero — não devem ser armazenados em cache sem justificativa de negócio documentada. A [ANPD](https://www.gov.br/anpd) orienta que dados pessoais sejam tratados com o princípio da necessidade e finalidade.

### Qual timeout configurar no SDK para a API de CPFHub.io?

A latência típica da API é de aproximadamente 900ms. Configure o timeout do cliente HTTP entre 10 e 30 segundos para absorver variações de rede sem prejudicar a experiência do usuário. Para o módulo de retry, use backoff exponencial com 3 tentativas máximas, aplicável apenas a erros de servidor (5xx) e timeouts — nunca a erros de autenticação (401).

## Leia também

- [Como validar CPF no frontend com React e API REST](https://cpfhub.io/blog/como-validar-cpf-no-frontend-com-react-e-api-rest)
- [Como criar um SDK próprio para a API de CPF do CPFHub](https://cpfhub.io/blog/como-criar-sdk-proprio-api-cpf-cpfhub)
- [Como consumir API de CPF em TypeScript com tipagem segura](https://cpfhub.io/blog/como-consumir-api-de-cpf-em-typescript-com-tipagem-segura)
- [Autenticação em APIs REST: como garantir segurança na consulta de CPF](https://cpfhub.io/blog/autenticacao-apis-rest-seguranca-consulta-cpf)

---

## Conclusão

Um SDK interno transforma uma integração fragmentada em um padrão unificado. Em vez de cada time reinventar a roda com sua própria implementação de retry, timeout e logging, todos consomem a mesma biblioteca testada e mantida centralmente.

A [**CPFHub.io**](https://www.cpfhub.io/) oferece uma API REST simples para consultas de CPF — endpoint único, autenticação por header e latência de ~900ms. Cadastre-se em [cpfhub.io](https://www.cpfhub.io/) e comece com 50 consultas gratuitas por mês.