Integrar uma consulta de CNPJ não deveria significar apenas montar uma URL e executar fetch. Em produção, a aplicação precisa validar o número antes da chamada, proteger o token, tratar limites, definir timeout e evitar pagar repetidamente pelo mesmo dado. Este tutorial implementa esse fluxo em Python e Node.js usando endpoints reais da EcoBigData.
A referência atualizada de autenticação, campos e exemplos fica na página da API de dados. Os exemplos abaixo usam:
GET /api/company/details?cnpj={cnpj}para os dados da empresa;GET /api/company/socios?cnpj={cnpj}para o quadro societário;- cabeçalho
Authorization: Bearer SEU_TOKEN.
Use CNPJs de teste autorizados e nunca grave o token no repositório.
Arquitetura recomendada
Um cliente de API CNPJ robusto pode ser dividido em cinco etapas:
- remover pontuação e validar os dígitos verificadores;
- procurar uma resposta recente no cache;
- chamar a API com timeout;
- classificar erros entre definitivos e transitórios;
- armazenar a resposta por um período compatível com o processo.
Essa separação evita chamadas desnecessárias e facilita trocar de provedor. Se você ainda está avaliando opções, veja o comparativo de APIs CNPJ.
Variáveis de ambiente
Defina a URL base e o token fora do código:
export ECOBIGDATA_API_URL="https://ecobigdata.com.br"
export ECOBIGDATA_TOKEN="seu_token_aqui"Em produção, use o gerenciador de segredos da nuvem ou da plataforma de deploy. Arquivos .env são convenientes no desenvolvimento, mas não devem ser enviados ao Git.
Validar o CNPJ antes da chamada
O CNPJ tradicional possui 14 dígitos. Os dois últimos são verificadores calculados por módulo 11. Validar localmente reduz respostas 400, consumo desnecessário e ruído nos logs.
Função pronta em Python
import re
def normalize_cnpj(value: str) -> str:
return re.sub(r"\D", "", value or "")
def is_valid_cnpj(value: str) -> bool:
cnpj = normalize_cnpj(value)
if len(cnpj) != 14 or len(set(cnpj)) == 1:
return False
def digit(base: str, weights: list[int]) -> str:
total = sum(int(number) * weight for number, weight in zip(base, weights))
remainder = total % 11
return "0" if remainder < 2 else str(11 - remainder)
first = digit(cnpj[:12], [5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2])
second = digit(cnpj[:12] + first, [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2])
return cnpj[-2:] == first + secondFunção pronta em Node.js
export function normalizeCnpj(value) {
return String(value ?? "").replace(/\D/g, "");
}
export function isValidCnpj(value) {
const cnpj = normalizeCnpj(value);
if (cnpj.length !== 14 || /^(\d)\1{13}$/.test(cnpj)) return false;
const calculate = (base, weights) => {
const total = [...base].reduce(
(sum, number, index) => sum + Number(number) * weights[index],
0,
);
const remainder = total % 11;
return remainder < 2 ? "0" : String(11 - remainder);
};
const first = calculate(cnpj.slice(0, 12), [5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]);
const second = calculate(cnpj.slice(0, 12) + first, [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]);
return cnpj.endsWith(first + second);
}O algoritmo evita a armadilha de aceitar sequências repetidas, como 00000000000000, que podem satisfazer implementações incompletas.
Cliente completo em Python
O exemplo usa requests, timeout explícito, cache em memória e retry exponencial para 429 e falhas transitórias. Em múltiplos processos, substitua o dicionário por Redis, banco ou cache compartilhado.
import os
import random
import time
from dataclasses import dataclass
from typing import Any
import requests
BASE_URL = os.environ.get("ECOBIGDATA_API_URL", "https://ecobigdata.com.br")
TOKEN = os.environ["ECOBIGDATA_TOKEN"]
@dataclass
class CacheEntry:
expires_at: float
value: dict[str, Any]
CACHE: dict[str, CacheEntry] = {}
CACHE_TTL_SECONDS = 6 * 60 * 60
class ApiError(RuntimeError):
pass
def cached_get(key: str):
entry = CACHE.get(key)
if not entry:
return None
if entry.expires_at <= time.time():
CACHE.pop(key, None)
return None
return entry.value
def cached_put(key: str, value: dict):
CACHE[key] = CacheEntry(time.time() + CACHE_TTL_SECONDS, value)
def request_company(path: str, cnpj_input: str, attempts: int = 4) -> dict:
cnpj = normalize_cnpj(cnpj_input)
if not is_valid_cnpj(cnpj):
raise ValueError("CNPJ inválido")
cache_key = f"{path}:{cnpj}"
if cached := cached_get(cache_key):
return cached
url = f"{BASE_URL}{path}"
headers = {
"Authorization": f"Bearer {TOKEN}",
"Accept": "application/json",
}
for attempt in range(attempts):
try:
response = requests.get(
url,
params={"cnpj": cnpj},
headers=headers,
timeout=(3.05, 20),
)
except (requests.Timeout, requests.ConnectionError) as error:
if attempt == attempts - 1:
raise ApiError(f"Falha de rede após {attempts} tentativas") from error
delay = (2 ** attempt) + random.random()
time.sleep(delay)
continue
if response.status_code == 200:
value = response.json()
cached_put(cache_key, value)
return value
if response.status_code == 401:
raise ApiError("Token ausente, expirado ou inválido")
if response.status_code == 403:
raise ApiError("Token sem permissão ou plano sem acesso ao recurso")
if response.status_code == 404:
raise ApiError("CNPJ não encontrado")
retryable = response.status_code == 429 or response.status_code >= 500
if not retryable or attempt == attempts - 1:
raise ApiError(
f"API retornou HTTP {response.status_code}: {response.text[:200]}"
)
retry_after = response.headers.get("Retry-After")
delay = float(retry_after) if retry_after and retry_after.isdigit() else 2 ** attempt
time.sleep(delay + random.random())
raise ApiError("Tentativas esgotadas")
def company_details(cnpj: str) -> dict:
return request_company("/api/company/details", cnpj)
def company_partners(cnpj: str) -> dict:
return request_company("/api/company/socios", cnpj)Uso:
details = company_details("00.000.000/0001-91")
partners = company_partners("00.000.000/0001-91")
print(details)
print(partners)O CNPJ do exemplo deve ser substituído por um número válido disponível para seu ambiente. Não use dados reais em testes automatizados sem controlar finalidade, logs e retenção.
Cliente completo em Node.js
O Node 18 ou superior oferece fetch nativo. O exemplo usa AbortSignal.timeout, cache em Map e respeita Retry-After quando o servidor envia o cabeçalho.
const BASE_URL = process.env.ECOBIGDATA_API_URL ?? "https://ecobigdata.com.br";
const TOKEN = process.env.ECOBIGDATA_TOKEN;
const CACHE_TTL_MS = 6 * 60 * 60 * 1000;
const cache = new Map();
if (!TOKEN) throw new Error("ECOBIGDATA_TOKEN não definido");
const sleep = (milliseconds) =>
new Promise((resolve) => setTimeout(resolve, milliseconds));
function getCached(key) {
const entry = cache.get(key);
if (!entry) return null;
if (entry.expiresAt <= Date.now()) {
cache.delete(key);
return null;
}
return entry.value;
}
function setCached(key, value) {
cache.set(key, { value, expiresAt: Date.now() + CACHE_TTL_MS });
}
async function requestCompany(path, cnpjInput, attempts = 4) {
const cnpj = normalizeCnpj(cnpjInput);
if (!isValidCnpj(cnpj)) throw new TypeError("CNPJ inválido");
const cacheKey = `${path}:${cnpj}`;
const cached = getCached(cacheKey);
if (cached) return cached;
const url = new URL(path, BASE_URL);
url.searchParams.set("cnpj", cnpj);
for (let attempt = 0; attempt < attempts; attempt += 1) {
let response;
try {
response = await fetch(url, {
headers: {
Authorization: `Bearer ${TOKEN}`,
Accept: "application/json",
},
signal: AbortSignal.timeout(20_000),
});
} catch (error) {
if (attempt === attempts - 1) {
throw new Error(`Falha de rede após ${attempts} tentativas`, { cause: error });
}
await sleep((2 ** attempt) * 1000 + Math.random() * 500);
continue;
}
if (response.ok) {
const value = await response.json();
setCached(cacheKey, value);
return value;
}
if (response.status === 401) throw new Error("Token inválido ou expirado");
if (response.status === 403) throw new Error("Acesso não permitido para este token");
if (response.status === 404) throw new Error("CNPJ não encontrado");
const retryable = response.status === 429 || response.status >= 500;
if (!retryable || attempt === attempts - 1) {
const body = (await response.text()).slice(0, 200);
throw new Error(`API retornou HTTP ${response.status}: ${body}`);
}
const retryAfter = Number(response.headers.get("retry-after"));
const delay = Number.isFinite(retryAfter) && retryAfter > 0
? retryAfter * 1000
: (2 ** attempt) * 1000;
await sleep(delay + Math.random() * 500);
}
throw new Error("Tentativas esgotadas");
}
export const companyDetails = (cnpj) =>
requestCompany("/api/company/details", cnpj);
export const companyPartners = (cnpj) =>
requestCompany("/api/company/socios", cnpj);Uso:
const details = await companyDetails("00.000.000/0001-91");
const partners = await companyPartners("00.000.000/0001-91");
console.log({ details, partners });Como tratar 401, 403 e 429
Esses códigos não devem receber o mesmo tratamento.
HTTP 401
Indica credencial ausente, inválida ou expirada. Repetir imediatamente com o mesmo token apenas aumenta tráfego. Atualize a credencial ou execute o fluxo de renovação antes de tentar novamente.
HTTP 403
O servidor reconheceu a credencial, mas negou o recurso. Verifique permissões, plano, escopo e situação da conta. Também não é um erro para retry automático.
HTTP 429
O limite foi atingido. Respeite Retry-After quando presente. Sem esse cabeçalho, aplique backoff exponencial com pequena aleatoriedade, chamada de jitter. A aleatoriedade impede que várias instâncias voltem ao mesmo tempo.
Erros 500, 502, 503, 504, timeout e falha de conexão podem ser transitórios. Restrinja o número de tentativas e registre a causa final. Retry infinito converte uma indisponibilidade parcial em sobrecarga.
Estratégia de cache por CNPJ
O TTL depende do uso:
- preenchimento de formulário: algumas horas podem ser suficientes;
- análise cadastral: registre quando o dado foi consultado;
- monitoramento: mantenha o snapshot e programe nova consulta;
- situação crítica: use a política de atualização contratada.
A chave deve incluir endpoint, CNPJ e, se aplicável, versão ou conjunto de campos. Não misture a resposta de details com a de socios. Para evitar stampede, um cache distribuído pode aplicar lock curto enquanto uma instância atualiza o item.
Dados cadastrais também têm implicações de privacidade e governança. Defina finalidade, usuários autorizados, prazo de retenção e informações que podem aparecer em logs.
Checklist para produção
- token somente no servidor;
- HTTPS obrigatório;
- validação do CNPJ antes da rede;
- timeout de conexão e resposta;
- retry apenas para erros transitórios;
- respeito ao
Retry-After; - cache com TTL e chave por endpoint;
- logs sem token e sem resposta completa desnecessária;
- métricas de latência, 429 e falhas;
- teste de contrato para detectar mudança de esquema.
Faça primeiro uma consulta CNPJ manual para conhecer os dados e só depois automatize os campos realmente necessários.
Perguntas frequentes
Posso chamar a API diretamente do navegador?
Não é recomendado quando a chamada exige um Bearer token secreto. O navegador expõe credenciais ao usuário e a extensões. Crie um endpoint no seu backend, aplique autorização e faça a chamada à EcoBigData no servidor.
Devo repetir uma chamada que retornou 401?
Não com a mesma credencial. Corrija ou renove o token. Retry automático é apropriado para 429, falhas de rede e alguns erros 5xx, sempre com limite.
Por quanto tempo devo manter o CNPJ em cache?
Não existe TTL universal. Considere a criticidade e a frequência de mudança do campo. Registre a data da consulta e use um TTL menor quando situação cadastral ou QSA precisarem de atualização frequente.
Como validar CNPJ alfanumérico?
Os exemplos implementam o formato numérico tradicional de 14 dígitos. A Receita Federal prevê CNPJ alfanumérico para novas inscrições a partir de 31 de julho de 2026. CNPJs atuais continuam válidos, mas a normalização e o cálculo devem seguir a nova especificação oficial; não force letras no algoritmo numérico antigo.
O endpoint de sócios substitui uma análise de KYB?
Não. Ele fornece dados para apoiar o processo. Beneficiário final, documentos, listas restritivas, contexto da operação e revisão humana podem continuar necessários conforme risco e obrigação aplicável.