Criar SDK interno para consultas de CPF padronizadas

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.

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:

Cliente HTTP: faz a chamada à API com timeout configurado.
Resiliência: retry com backoff exponencial e circuit breaker.
Observabilidade: logging estruturado e métricas.
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

# 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

# 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

# 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

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

// 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 define como comunicar compatibilidade entre versões de forma clara.

Testes do SDK

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 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).

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 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 e comece com 50 consultas gratuitas por mês.

CPFHub.io

Pronto para integrar a API?

50 consultas gratuitas para testar agora. Sem cartão de crédito. Acesso imediato à documentação.

Começar grátis Ver documentação

Sobre a redação

Redação CPFHub.io

Time editorial especializado em APIs de CPF, identidade digital e compliance no mercado brasileiro. Produzimos guias técnicos, análises regulatórias e tutoriais sobre LGPD e KYC para desenvolvedores e líderes de produto.

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