Integrando SmartCPF em Google Cloud Functions com Python

Validar e consultar CPF em escala é um requisito comum em sistemas de onboarding, prevenção a fraude e compliance. A API SmartCPF resolve isso expondo um endpoint REST simples que retorna dados cadastrais, situação cadastral e informações de contato — tudo sob um modelo de créditos previsível. Este artigo mostra como consumir a API a partir de uma Google Cloud Function escrita em Python, cobrindo autenticação, tratamento de erros e deploy.

O cenário típico: uma função HTTP recebe um CPF do frontend, chama a SmartCPF, normaliza a resposta e devolve apenas os campos necessários à aplicação. Manter a lógica de integração isolada em uma Function evita acoplamento no backend principal e permite escalar independentemente conforme o volume de consultas.

Como funciona a integração com Cloud Functions

Google Cloud Functions executa código em resposta a eventos HTTP (ou Pub/Sub, Storage, etc.) sem gerenciar servidores. Para a SmartCPF, o fluxo é direto: a Function recebe a requisição, valida o CPF, adiciona o header X-API-KEY, faz a chamada HTTPS para https://api.smartcpf.com.br/v2/individuals/{cpf}?package=X, trata códigos de status e erros de rede, e retorna JSON padronizado ao chamador. O segredo da API Key fica em Secret Manager, nunca no código.

Vantagens de usar Cloud Functions para esta integração

1. Isolamento de responsabilidades
A lógica de consulta a terceiros fica desacoplada do monólito, facilitando testes e trocas de provedor.

2. Escala automática
Picos de onboarding (ex: campanhas de marketing) não derrubam a aplicação principal; a Function escala horizontalmente.

3. Custo por execução
Você paga apenas pelas invocações reais — ideal para volumes variáveis de consultas de CPF.

4. Secrets nativos
Integração direta com Secret Manager elimita variáveis de ambiente hardcoded e rotação manual de chaves.

5. Observabilidade integrada
Cloud Logging e Cloud Trace capturam latência, erros e volume sem instrumentação extra.

6. Deploy rápido
gcloud functions deploy coloca a nova versão em produção em segundos, com rollback trivial.

Desafios ao usar Cloud Functions para esta integração

1. Cold starts
A primeira invocação após inatividade adiciona latência (100–500 ms). Mitigue com min_instances=1 se latência consistente for crítica.

2. Tempo limite de execução
O padrão é 60 s (máx. 540 s). A SmartCPF responde em < 200 ms, mas timeouts de rede ou retries podem estourar o limite se não houver timeout no cliente HTTP.

3. Limites de concorrência por instância
Uma instância processa uma requisição por vez (exceto se configurado para concorrência). Alto volume exige mais instâncias, o que aumenta custo.

4. Gerenciamento de dependências
O requirements.txt deve ser leve; bibliotecas pesadas aumentam cold start e tempo de deploy.

5. VPC Connector para Secret Manager
Se a Function roda em rede VPC, acessar Secret Manager exige Serverless VPC Access Connector — custo e configuração extras.

Como implementar com a SmartCPF

Abaixo, uma Function HTTP pronta para deploy. Ela lê a API Key do Secret Manager no startup (cache em variável global), valida o CPF recebido, chama o endpoint com o header X-API-KEY, trata erros HTTP e de rede, e retorna apenas os campos essenciais normalizados.

main.py

"""
Google Cloud Function HTTP para consultar CPF na API SmartCPF v2.

Requisitos:
- Python 3.11+
- google-cloud-secret-manager
- functions-framework
- requests

Deploy:
gcloud functions deploy consultar-cpf \
  --gen2 --runtime=python311 --region=southamerica-east1 \
  --trigger-http --allow-unauthenticated \
  --entry-point=consultar_cpf \
  --set-secrets=SMARTCPF_API_KEY=projects/<PROJECT_ID>/secrets/smartcpf-api-key/versions/latest \
  --min-instances=1 --max-instances=10 --timeout=30s --memory=256MiB
"""

import os
import re
import json
import logging
from typing import Any, Dict, Tuple

import requests
from google.cloud import secretmanager

# Configuração de logging estruturado para Cloud Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Constantes da API SmartCPF
SMARTCPF_BASE_URL = "https://api.smartcpf.com.br/v2/individuals"
DEFAULT_PACKAGE = "X"  # Ajuste conforme seu plano
REQUEST_TIMEOUT = 10  # segundos (connect + read)

# Cache da API Key em variável global (persiste entre invocações quentes)
_API_KEY: str | None = None


def _get_api_key() -> str:
    """Obtém a API Key do Secret Manager (apenas na primeira invocação fria)."""
    global _API_KEY
    if _API_KEY is not None:
        return _API_KEY

    project_id = os.getenv("GCP_PROJECT") or os.getenv("GOOGLE_CLOUD_PROJECT")
    secret_name = os.getenv("SMARTCPF_API_KEY_SECRET", "smartcpf-api-key")
    version = os.getenv("SMARTCPF_API_KEY_VERSION", "latest")

    if not project_id:
        raise RuntimeError("GCP_PROJECT não definido no ambiente")

    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project_id}/secrets/{secret_name}/versions/{version}"
    response = client.access_secret_version(request={"name": name})
    _API_KEY = response.payload.data.decode("UTF-8").strip()
    logger.info("API Key carregada do Secret Manager")
    return _API_KEY


def _validar_cpf(cpf: str) -> str:
    """Remove formatação e valida dígitos verificadores do CPF."""
    numeros = re.sub(r"\D", "", cpf)
    if len(numeros) != 11 or numeros == numeros[0] * 11:
        raise ValueError("CPF inválido")

    # Cálculo do 1º dígito verificador
    soma = sum(int(numeros[i]) * (10 - i) for i in range(9))
    digito1 = (soma * 10 % 11) % 10
    # Cálculo do 2º dígito verificador
    soma = sum(int(numeros[i]) * (11 - i) for i in range(10))
    digito2 = (soma * 10 % 11) % 10

    if int(numeros[9]) != digito1 or int(numeros[10]) != digito2:
        raise ValueError("CPF inválido (dígitos verificadores)")

    return numeros


def _chamar_smartcpf(cpf: str, api_key: str) -> Tuple[Dict[str, Any], int]:
    """Faz a requisição HTTP para a API SmartCPF."""
    url = f"{SMARTCPF_BASE_URL}/{cpf}?package={DEFAULT_PACKAGE}"
    headers = {
        "X-API-KEY": api_key,  # Autenticação obrigatória via header
        "Accept": "application/json",
        "User-Agent": "smartcpf-cloud-function/1.0",
    }

    try:
        resp = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
        resp.raise_for_status()
        return resp.json(), resp.status_code
    except requests.exceptions.HTTPError as exc:
        status = exc.response.status_code if exc.response else 502
        logger.error("Erro HTTP SmartCPF: %s - %s", status, exc)
        return {"error": "Erro na API SmartCPF", "status": status}, status
    except requests.exceptions.Timeout:
        logger.error("Timeout ao consultar SmartCPF")
        return {"error": "Timeout na consulta", "status": 504}, 504
    except requests.exceptions.RequestException as exc:
        logger.exception("Erro de rede ao chamar SmartCPF")
        return {"error": "Falha de comunicação", "status": 502}, 502


def _normalizar_resposta(data: Dict[str, Any]) -> Dict[str, Any]:
    """Extrai apenas os campos relevantes da resposta da SmartCPF."""
    meta = data.get("meta", {})
    result = data.get("result", {})

    return {
        "consultId": meta.get("consultId"),
        "balance": meta.get("balance"),
        "executionTimeMs": meta.get("executionTimeMs"),
        "cpf": result.get("cpf"),
        "name": result.get("name"),
        "birthDate": result.get("birthDate"),
        "age": result.get("age"),
        "status": result.get("status"),
        "gender": result.get("gender"),
        "motherName": result.get("motherName"),
        "emails": result.get("emails", []),
        "phones": result.get("phones", []),
        "addresses": result.get("addresses", []),
    }


# --- Entry point da Cloud Function ---
def consultar_cpf(request) -> Tuple[str, int, Dict[str, str]]:
    """
    Cloud Function HTTP entry point.

    Espera JSON: {"cpf": "111.111.111-11"}
    Retorna JSON normalizado ou erro com status apropriado.
    """
    # CORS simples para chamadas de frontend
    headers = {
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "POST, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type",
        "Content-Type": "application/json; charset=utf-8",
    }

    if request.method == "OPTIONS":
        return ("", 204, headers)

    if request.method != "POST":
        return (json.dumps({"error": "Método não permitido"}), 405, headers)

    # Parse do body
    try:
        payload = request.get_json(silent=True) or {}
        cpf_raw = payload.get("cpf", "").strip()
    except Exception:
        return (json.dumps({"error": "Body JSON inválido"}), 400, headers)

    if not cpf_raw:
        return (json.dumps({"error": "Campo 'cpf' é obrigatório"}), 400, headers)

    # Validação de CPF
    try:
        cpf = _validar_cpf(cpf_raw)
    except ValueError as exc:
        return (json.dumps({"error": str(exc)}), 422, headers)

    # Obter API Key e consultar
    try:
        api_key = _get_api_key()
    except Exception as exc:
        logger.exception("Falha ao obter API Key")
        return (json.dumps({"error": "Erro de configuração interna"}), 500, headers)

    data, status = _chamar_smartcpf(cpf, api_key)

    if status != 200:
        return (json.dumps(data, ensure_ascii=False), status, headers)

    # Sucesso: normalizar e retornar
    normalized = _normalizar_resposta(data)
    return (json.dumps(normalized, ensure_ascii=False), 200, headers)

Resposta da API

A chamada bem-sucedida retorna a estrutura abaixo. Os campos mais úteis para a maioria das integrações são consultId (rastreabilidade), balance (créditos restantes), executionTimeMs (latência real) e o objeto result com dados do titular.

response.json

{
  "meta": {
    "searchType": "CPF G",
    "consultId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "user": "Fulano de Tal",
    "message": "CPF encontrado com sucesso",
    "ip": "127.0.0.1",
    "success": true,
    "apiVersion": "v2",
    "date": "2026-08-17T14:32:10.123Z",
    "executionTimeMs": 74.1,
    "balance": 9.77
  },
  "result": {
    "cpf": "111.111.111-11",
    "name": "João da Silva",
    "birthDate": "10/09/1992",
    "age": 33,
    "motherName": "Maria da Silva",
    "gender": "M",
    "status": "Regular",
    "emails": ["[email protected]"],
    "phones": [
      {
        "number": "(55) 99999-9999",
        "blockedTelemarketing": false,
        "operator": "VIVO",
        "type": "CELULAR",
        "whatsapp": false
      }
    ],
    "addresses": []
  },
  "success": true,
  "errors": []
}

O fluxo demonstrado resolve na prática:

Rastreabilidade: consultId permite auditar cada consulta nos logs da SmartCPF.
Controle de custo: balance mostra créditos restantes em tempo real — alerte quando < 10.
Observabilidade: executionTimeMs expõe a latência real da API para dashboards.
Dados prontos: CPF formatado, nome, idade, status cadastral, e-mails e telefones já normalizados.
Resiliência: Timeouts e erros HTTP são convertidos em códigos de status semânticos (502, 504, 422).

Por que usar a SmartCPF em produção

Latência consistente: P95 < 200 ms na região Sudeste, com SLA de disponibilidade 99,9%.
Integração direta: Um único endpoint, autenticação por header, sem OAuth flows ou SDKs pesados.
Modelo de créditos transparente: Você compra pacotes (ex: 1.000 consultas) e consome sob demanda — sem assinatura mensal.
Escalabilidade nativa: A API suporta milhares de RPS; sua Cloud Function escala junto automaticamente.
Dados enriquecidos: Além do status cadastral, retorna telefones com operadora, WhatsApp, e-mails e endereços quando disponíveis.

Conclusão

Isolar a consulta de CPF em uma Cloud Function mantém seu backend limpo, escalável e observável. O exemplo acima cobre validação local, autenticação segura via Secret Manager, timeouts defensivos e normalização de resposta — tudo em ~120 linhas. Crie sua conta na SmartCPF e faça sua primeira consulta em minutos.