# Como consumir API de CPF em Google Cloud Functions com Python

> Aprenda a criar uma Cloud Function em Python que consome a API de CPF da CPFHub.io com tratamento de erros, cache e deploy automatizado.

**Publicado:** 19/09/2026
**Autor:** Redação CPFHub.io
**URL:** https://www.cpfhub.io/blog/como-consumir-api-cpf-google-cloud-functions-python

---


Google Cloud Functions é a plataforma serverless do Google que executa código Python em resposta a eventos HTTP sem gerenciar servidores. Integrar a API da CPFHub.io nesse ambiente permite criar endpoints de validação de CPF que escalam automaticamente, com resposta em torno de ~300ms e custo proporcional ao uso real. A autenticação usa o header `x-api-key` e o endpoint segue o padrão `GET https://api.cpfhub.io/cpf/{CPF_NUMBER}`.

---

## Estrutura do projeto

Uma Cloud Function em Python requer no mínimo dois arquivos:

```
cpf-validator/
 main.py # Funcao principal
 requirements.txt # Dependencias
```

Para projetos mais complexos, podemos adicionar módulos auxiliares:

```
cpf-validator/
 main.py
 cpfhub_service.py
 validators.py
 requirements.txt
```

---

## Serviço de consulta de CPF

Primeiro, criamos o módulo que encapsula a comunicação com a API da [**CPFHub.io**](https://www.cpfhub.io/)

```python
# cpfhub_service.py
import requests
import os
import re
from google.cloud import secretmanager

TIMEOUT_SECONDS = 10
API_URL = "https://api.cpfhub.io/cpf"

_cached_api_key = None

def get_api_key():
 """Recupera a chave de API do Secret Manager com cache."""
 global _cached_api_key

 if _cached_api_key:
 return _cached_api_key

 # Tentar variavel de ambiente primeiro (desenvolvimento local)
 api_key = os.environ.get("CPFHUB_API_KEY")

 if not api_key:
 # Buscar no Secret Manager (producao)
 client = secretmanager.SecretManagerServiceClient()
 project_id = os.environ.get("GCP_PROJECT")
 secret_name = f"projects/{project_id}/secrets/cpfhub-api-key/versions/latest"

 response = client.access_secret_version(name=secret_name)
 api_key = response.payload.data.decode("utf-8")

 _cached_api_key = api_key
 return api_key

def limpar_cpf(cpf_raw):
 """Remove caracteres nao numericos do CPF."""
 return re.sub(r"\D", "", cpf_raw)

def validar_formato_cpf(cpf):
 """Valida formato basico do CPF."""
 if len(cpf) != 11:
 return False, "CPF deve conter 11 digitos"

 if len(set(cpf)) == 1:
 return False, "CPF invalido (todos os digitos iguais)"

 return True, None

def consultar_cpf(cpf_numero):
 """Consulta CPF na API CPFHub.io."""
 cpf_limpo = limpar_cpf(cpf_numero)

 valido, erro = validar_formato_cpf(cpf_limpo)
 if not valido:
 return {"sucesso": False, "erro": erro}

 api_key = get_api_key()

 headers = {
 "x-api-key": api_key,
 "Accept": "application/json"
 }

 try:
 response = requests.get(
 f"{API_URL}/{cpf_limpo}",
 headers=headers,
 timeout=TIMEOUT_SECONDS
 )
 response.raise_for_status()

 dados = response.json()

 if not dados.get("success"):
 return {"sucesso": False, "erro": "CPF nao encontrado"}

 return {
 "sucesso": True,
 "dados": {
 "cpf": dados["data"]["cpf"],
 "nome": dados["data"]["name"],
 "genero": dados["data"]["gender"],
 "data_nascimento": dados["data"]["birthDate"],
 "dia": dados["data"]["day"],
 "mes": dados["data"]["month"],
 "ano": dados["data"]["year"]
 }
 }

 except requests.Timeout:
 return {"sucesso": False, "erro": "Timeout na consulta de CPF"}
 except requests.ConnectionError:
 return {"sucesso": False, "erro": "Erro de conexao com a API"}
 except requests.HTTPError as e:
 return {"sucesso": False, "erro": f"Erro HTTP: {e.response.status_code}"}
 except Exception as e:
 return {"sucesso": False, "erro": f"Erro inesperado: {str(e)}"}
```

---

## Função principal (HTTP trigger)

A Cloud Function HTTP recebe um objeto `Request` do Flask e retorna uma resposta:

```python
# main.py
import functions_framework
from flask import jsonify, make_response
from cpfhub_service import consultar_cpf, limpar_cpf

CORS_HEADERS = {
 "Access-Control-Allow-Origin": "*",
 "Access-Control-Allow-Methods": "GET, OPTIONS",
 "Access-Control-Allow-Headers": "Content-Type, Authorization",
 "Access-Control-Max-Age": "3600"
}

def criar_resposta(dados, status_code=200):
 """Cria resposta HTTP com headers CORS."""
 response = make_response(jsonify(dados), status_code)
 for key, value in CORS_HEADERS.items():
 response.headers[key] = value
 return response

@functions_framework.http
def validar_cpf(request):
 """Cloud Function para validacao de CPF.

 Args:
 request: Objeto Flask Request.

 Returns:
 Resposta JSON com dados do CPF ou mensagem de erro.
 """
 # Tratar preflight CORS
 if request.method == "OPTIONS":
 return criar_resposta({})

 if request.method != "GET":
 return criar_resposta(
 {"erro": "Metodo nao permitido. Use GET."},
 405
 )

 # Extrair CPF da URL ou query string
 cpf = None

 # Tentar extrair do path: /cpf/12345678901
 path = request.path.strip("/")
 partes = path.split("/")
 if len(partes) >= 2 and partes[0] == "cpf":
 cpf = partes[1]

 # Fallback: query string ?cpf=12345678901
 if not cpf:
 cpf = request.args.get("cpf")

 if not cpf:
 return criar_resposta(
 {"erro": "CPF e obrigatorio. Use /cpf/{numero} ou ?cpf={numero}"},
 400
 )

 cpf_limpo = limpar_cpf(cpf)

 if len(cpf_limpo) != 11:
 return criar_resposta(
 {"erro": "CPF deve conter 11 digitos"},
 400
 )

 resultado = consultar_cpf(cpf_limpo)

 if not resultado["sucesso"]:
 status = 404 if "nao encontrado" in resultado.get("erro", "") else 500
 return criar_resposta({"erro": resultado["erro"]}, status)

 return criar_resposta({
 "valido": True,
 "dados": resultado["dados"]
 })
```

---

## Dependências e configuração

O arquivo `requirements.txt`:

```
functions-framework==3.*
requests>=2.31.0,<3.0.0
google-cloud-secret-manager>=2.16.0,<3.0.0
```

---

## Deploy da Cloud Function

Deploy usando o Google Cloud CLI. Consulte a [documentação oficial do Google Cloud Functions](https://cloud.google.com/functions/docs) para detalhes sobre flags e configurações avançadas:

```bash
gcloud functions deploy validar-cpf \
 --gen2 \
 --runtime=python312 \
 --region=southamerica-east1 \
 --source=. \
 --entry-point=validar_cpf \
 --trigger-http \
 --allow-unauthenticated \
 --memory=256MB \
 --timeout=30s \
 --set-env-vars="GCP_PROJECT=meu-projeto"
```

A região `southamerica-east1` (São Paulo) minimiza a latência para usuários brasileiros. Combinada com o tempo de resposta de ~300ms da API da CPFHub.io, a resposta total fica bem abaixo de 1 segundo na maioria dos cenários.

---

## Gerenciamento de segredos

Para armazenar a chave de API de forma segura no Google Cloud:

```bash
# Criar o segredo
echo -n "sua_chave_cpfhub" | gcloud secrets create cpfhub-api-key \
 --data-file=- \
 --replication-policy=user-managed \
 --locations=southamerica-east1

# Conceder acesso a Cloud Function
gcloud secrets add-iam-policy-binding cpfhub-api-key \
 --member="serviceAccount:meu-projeto@appspot.gserviceaccount.com" \
 --role="roles/secretmanager.secretAccessor"
```

---

## Testes locais

O Functions Framework permite testar localmente antes do deploy:

```bash
# Instalar dependencias
pip install -r requirements.txt

# Definir variavel de ambiente para teste local
export CPFHUB_API_KEY="sua_chave_de_teste"

# Iniciar servidor local
functions-framework --target=validar_cpf --port=8080
```

Teste com curl:

```bash
curl "http://localhost:8080/cpf/12345678901"
```

---

## Cache com Memorystore ou Firestore

Para reduzir chamadas à API da CPFHub.io e melhorar a performance, implemente cache usando Firestore:

```python
from google.cloud import firestore
from datetime import datetime, timedelta

db = firestore.Client()
CACHE_TTL_HOURS = 24

def buscar_cache(cpf):
 """Busca resultado em cache no Firestore."""
 doc_ref = db.collection("cpf_cache").document(cpf)
 doc = doc_ref.get()

 if doc.exists:
 dados = doc.to_dict()
 expiracao = dados.get("expira_em")

 if expiracao and expiracao > datetime.utcnow():
 return dados.get("resultado")

 return None

def salvar_cache(cpf, resultado):
 """Salva resultado no cache do Firestore."""
 doc_ref = db.collection("cpf_cache").document(cpf)
 doc_ref.set({
 "resultado": resultado,
 "expira_em": datetime.utcnow() + timedelta(hours=CACHE_TTL_HOURS),
 "atualizado_em": datetime.utcnow()
 })
```

---

## Monitoramento e alertas

Configure alertas no Cloud Monitoring para acompanhar a saúde da integração:

* **Taxa de erro** — alerte quando a taxa de erros 5xx ultrapassar 5% em 5 minutos.
* **Latência** — alerte quando o p99 de latência ultrapassar 2 segundos.
* **Invocações** — monitore o volume para prever quando migrar para o plano Pro da CPFHub.io.
* **Cold starts** — acompanhe a frequência para avaliar se Provisioned Instances são necessárias.

---

## Perguntas frequentes

### Como funciona o rate limit da API CPFHub.io no plano gratuito?

No plano gratuito, o rate limit é de 1 requisição a cada 2 segundos. Ultrapassar essa taxa retorna HTTP 429 — mas isso é diferente de esgotar a cota mensal. Se você consumir as 50 consultas do plano gratuito, a API não bloqueia: cada consulta extra é cobrada a R$0,15. Para evitar surpresas, monitore o consumo no painel e considere implementar cache local.

### Cloud Functions é uma boa opção para validação de CPF em alta frequência?

Sim, especialmente quando combinada com cache no Firestore. O modelo serverless escala automaticamente com a demanda, e o cache elimina chamadas repetidas para o mesmo CPF. Para volumes muito altos, o plano Pro da CPFHub.io (R$149/mês, 1.000 consultas incluídas) oferece rate limit de 1 req/s — o dobro do plano gratuito.

### É seguro armazenar a chave de API em variáveis de ambiente no Cloud Functions?

Variáveis de ambiente no Cloud Functions são uma solução aceitável para desenvolvimento, mas para produção o recomendado é o Google Secret Manager. Ele criptografa o segredo em repouso, controla o acesso via IAM e registra cada acesso em audit logs — atendendo às boas práticas de segurança e aos requisitos de conformidade com a LGPD.

### O que acontece se a API da CPFHub.io ficar indisponível durante a execução da função?

O código usa `requests.Timeout` e `requests.ConnectionError` para capturar essas situações e retornar mensagens de erro claras. Uma estratégia complementar é implementar retry com backoff exponencial usando a biblioteca `tenacity`, garantindo resiliência sem sobrecarregar a API em momentos de instabilidade.

### 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)
- [API de CPF grátis para desenvolvedores: como começar em 5 minutos](https://cpfhub.io/blog/api-cpf-gratis-desenvolvedores-comecar-5-minutos)
- [SLA de API de CPF: níveis de disponibilidade](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade)
- [Diferença entre validação de CPF e consulta de CPF: quando usar cada uma](https://cpfhub.io/blog/diferenca-entre-validacao-de-cpf-e-consulta-de-cpf-quando-usar-cada-uma)

---

## Conclusão

Google Cloud Functions com Python oferecem uma base sólida para validação de CPF em ambiente serverless. A integração com Secret Manager resolve a questão de segurança das credenciais, o Functions Framework simplifica os testes locais, e a escolha da região São Paulo reduz a latência para usuários brasileiros. Com cache no Firestore e monitoramento ativo, você tem uma solução resiliente que escala sem esforço de operação.

Se você ainda não testou a API da [**CPFHub.io**](https://www.cpfhub.io/), o plano gratuito oferece 50 consultas por mês sem cartão de crédito — tempo suficiente para validar a integração completa. Crie sua conta em [cpfhub.io](https://www.cpfhub.io/) e comece a consultar em menos de 30 minutos.