# API de CPF: como monitorar consumo e custos com observabilidade

> Aprenda a monitorar consumo, custos e saúde da integração com API de CPF usando métricas, logs e alertas de observabilidade.

**Publicado:** 21/08/2026
**Autor:** Redação CPFHub.io
**URL:** https://cpfhub.io/blog/api-cpf-monitorar-consumo-custos-observabilidade

---


Monitorar o consumo da API de CPF é essencial para evitar surpresas de cobrança e garantir que a integração funcione dentro do esperado. A CPFHub.io não bloqueia quando a cota é esgotada -- cobra R$0,15 por consulta extra -- então sem visibilidade em tempo real, o volume pode escapar antes que qualquer alerta dispare. Este guia mostra como implementar métricas, alertas e dashboards para manter o controle completo da integração.

---

## As três métricas essenciais

Para monitorar uma integração com API de CPF, concentre-se nestas três métricas:

### 1. Consumo de cota

Quantas consultas foram realizadas no período atual. Essencial para evitar surpresas de cobrança ou interrupção de serviço.

### 2. Latência de resposta

Tempo entre enviar a requisição e receber a resposta. A API da CPFHub.io tem latência média de aproximadamente 900ms -- valores consistentemente acima de 2 segundos indicam degradação que merece investigação.

### 3. Taxa de erro

Porcentagem de requisições que falharam (timeouts, erros 4xx/5xx). Um aumento repentino exige investigação imediata.

---

## Implementando um wrapper com métricas

O primeiro passo é encapsular as chamadas à API em um wrapper que registra métricas automaticamente:

```python
import time
import requests
import logging
from dataclasses import dataclass, field
from typing import Optional, Dict, List
from datetime import datetime, date

logger = logging.getLogger(__name__)

@dataclass
class MetricasAPI:
 """Armazena métricas de uso da API."""
 total_requisicoes: int = 0
 requisicoes_sucesso: int = 0
 requisicoes_erro: int = 0
 timeouts: int = 0
 latencias: List[float] = field(default_factory=list)
 consumo_diario: Dict[str, int] = field(default_factory=dict)

 @property
 def taxa_erro(self) -> float:
 if self.total_requisicoes == 0:
 return 0.0
 return (self.requisicoes_erro / self.total_requisicoes) * 100

 @property
 def latencia_media(self) -> float:
 if not self.latencias:
 return 0.0
 return sum(self.latencias) / len(self.latencias)

 @property
 def latencia_p95(self) -> float:
 if not self.latencias:
 return 0.0
 sorted_lat = sorted(self.latencias)
 idx = int(len(sorted_lat) * 0.95)
 return sorted_lat[idx]

 def registrar_consumo_hoje(self):
 hoje = date.today().isoformat()
 self.consumo_diario[hoje] = self.consumo_diario.get(hoje, 0) + 1

 def consumo_mes_atual(self) -> int:
 mes_atual = date.today().strftime("%Y-%m")
 return sum(
 v for k, v in self.consumo_diario.items()
 if k.startswith(mes_atual)
 )

class CPFClientMonitorado:
 """Cliente de API de CPF com monitoramento integrado."""

 BASE_URL = "https://api.cpfhub.io/cpf"
 TIMEOUT = 30

 def __init__(self, api_key: str, cota_mensal: int = 1000):
 self.api_key = api_key
 self.cota_mensal = cota_mensal
 self.metricas = MetricasAPI()
 self.session = requests.Session()
 self.session.headers.update({
 "x-api-key": self.api_key,
 "Accept": "application/json"
 })

 def consultar(self, cpf: str) -> Optional[Dict]:
 """Consulta CPF e registra métricas."""
 self.metricas.total_requisicoes += 1
 self.metricas.registrar_consumo_hoje()

 # Alerta de cota
 consumo = self.metricas.consumo_mes_atual()
 if consumo >= self.cota_mensal * 0.8:
 logger.warning(
 f"ALERTA: {consumo}/{self.cota_mensal} consultas usadas "
 f"({(consumo/self.cota_mensal)*100:.0f}%)"
 )

 inicio = time.time()

 try:
 response = self.session.get(
 f"{self.BASE_URL}/{cpf}",
 timeout=self.TIMEOUT
 )
 latencia = time.time() - inicio
 self.metricas.latencias.append(latencia)

 response.raise_for_status()
 dados = response.json()

 if dados.get("success"):
 self.metricas.requisicoes_sucesso += 1
 return dados
 else:
 self.metricas.requisicoes_erro += 1
 return None

 except requests.exceptions.Timeout:
 self.metricas.requisicoes_erro += 1
 self.metricas.timeouts += 1
 logger.error(f"Timeout ao consultar CPF")
 return None

 except requests.exceptions.RequestException as e:
 latencia = time.time() - inicio
 self.metricas.latencias.append(latencia)
 self.metricas.requisicoes_erro += 1
 logger.error(f"Erro na requisicao: {e}")
 return None

 def relatorio(self) -> str:
 """Gera relatório de métricas."""
 m = self.metricas
 return (
 f"=== Relatorio de Metricas ===\n"
 f"Total de requisicoes: {m.total_requisicoes}\n"
 f"Sucesso: {m.requisicoes_sucesso}\n"
 f"Erros: {m.requisicoes_erro}\n"
 f"Timeouts: {m.timeouts}\n"
 f"Taxa de erro: {m.taxa_erro:.1f}%\n"
 f"Latencia media: {m.latencia_media*1000:.0f}ms\n"
 f"Latencia P95: {m.latencia_p95*1000:.0f}ms\n"
 f"Consumo mensal: {m.consumo_mes_atual()}/{self.cota_mensal}\n"
 )
```

---

## Configurando alertas

Alertas proativos evitam que problemas sejam descobertos tarde demais:

```python
class AlertaConsumo:
 """Sistema de alertas baseado em métricas."""

 def __init__(self, cliente: CPFClientMonitorado):
 self.cliente = cliente

 def verificar_alertas(self) -> List[str]:
 """Retorna lista de alertas ativos."""
 alertas = []
 m = self.cliente.metricas

 # Alerta de cota
 consumo = m.consumo_mes_atual()
 percentual = (consumo / self.cliente.cota_mensal) * 100
 if percentual >= 90:
 alertas.append(f"CRITICO: Cota em {percentual:.0f}% ({consumo}/{self.cliente.cota_mensal})")
 elif percentual >= 80:
 alertas.append(f"AVISO: Cota em {percentual:.0f}% ({consumo}/{self.cliente.cota_mensal})")

 # Alerta de latência
 if m.latencia_media > 2.0:
 alertas.append(f"AVISO: Latencia media alta ({m.latencia_media*1000:.0f}ms)")

 # Alerta de taxa de erro
 if m.taxa_erro > 5.0:
 alertas.append(f"CRITICO: Taxa de erro em {m.taxa_erro:.1f}%")
 elif m.taxa_erro > 2.0:
 alertas.append(f"AVISO: Taxa de erro em {m.taxa_erro:.1f}%")

 return alertas
```

---

## Exportando métricas para Prometheus

Se a sua equipe usa [Prometheus](https://prometheus.io/docs/introduction/overview/) e Grafana, exporte as métricas no formato padrão:

```python
from prometheus_client import Counter, Histogram, Gauge

# Definir métricas
cpf_requests_total = Counter(
 "cpf_api_requests_total",
 "Total de requisicoes a API de CPF",
 ["status"]
)

cpf_request_duration = Histogram(
 "cpf_api_request_duration_seconds",
 "Latencia das requisicoes a API de CPF",
 buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)

cpf_quota_usage = Gauge(
 "cpf_api_quota_usage",
 "Consumo atual da cota mensal"
)
```

---

## Monitorando via cURL

Para verificações rápidas sem código adicional, use cURL para medir latência:

```bash
curl -o /dev/null -s -w "Status: %{http_code}\nDNS: %{time_namelookup}s\nConexao: %{time_connect}s\nTLS: %{time_appconnect}s\nTransferencia: %{time_starttransfer}s\nTotal: %{time_total}s\n" \
 "https://api.cpfhub.io/cpf/12345678900" \
 -H "x-api-key: SUA_CHAVE_API" \
 -H "Accept: application/json" \
 --connect-timeout 10 \
 --max-time 30
```

Saída esperada:

```
Status: 200
DNS: 0.023s
Conexao: 0.045s
TLS: 0.112s
Transferencia: 0.934s
Total: 0.935s
```

---

## Dashboards recomendados

Organize as métricas em três dashboards:

### Dashboard operacional

- Requisições por minuto (taxa de throughput).
- Latência média, P50, P95 e P99.
- Taxa de erro por tipo (timeout, 4xx, 5xx).

### Dashboard de consumo

- Consultas realizadas vs. cota disponível.
- Projeção de consumo para o restante do mês.
- Consumo por dia da semana (identifica padrões).

### Dashboard de custos

- Custo por consulta realizada.
- Custo acumulado no mês.
- Comparativo com meses anteriores.

---

## Logs estruturados

Além de métricas, mantenha logs estruturados para debugging:

```python
import json

def log_requisicao(cpf_masked, status_code, latencia, sucesso):
 """Registra log estruturado de uma requisição."""
 log_entry = {
 "timestamp": datetime.utcnow().isoformat(),
 "service": "cpf-api",
 "cpf_masked": cpf_masked,
 "status_code": status_code,
 "latency_ms": round(latencia * 1000),
 "success": sucesso
 }
 logger.info(json.dumps(log_entry))

# Exemplo de saída
# {"timestamp": "2026-08-21T14:30:00", "service": "cpf-api",
# "cpf_masked": "123***00", "status_code": 200,
# "latency_ms": 892, "success": true}
```

Note que o CPF é mascarado no log para conformidade com a LGPD.

---

## Automatizando relatórios semanais

```python
def gerar_relatorio_semanal(cliente: CPFClientMonitorado) -> str:
 """Gera relatório semanal de consumo."""
 m = cliente.metricas
 consumo = m.consumo_mes_atual()
 percentual = (consumo / cliente.cota_mensal) * 100

 return (
 f"Relatorio Semanal - API de CPF\n"
 f"Periodo: ultima semana\n"
 f"Consultas realizadas: {m.total_requisicoes}\n"
 f"Taxa de sucesso: {100 - m.taxa_erro:.1f}%\n"
 f"Latencia media: {m.latencia_media*1000:.0f}ms\n"
 f"Consumo mensal: {consumo}/{cliente.cota_mensal} ({percentual:.0f}%)\n"
 f"Projecao mensal: ~{int(consumo / max(date.today().day, 1) * 30)} consultas\n"
 )
```

---

## Perguntas frequentes

### Qual é a latência média da API de CPF e como monitorá-la?
A latência média da API da CPFHub.io é de aproximadamente 900ms. Monitore o P95 (percentil 95) além da média -- uma média saudável pode esconder picos que afetam a experiência do usuário. Configure um alerta quando o P95 ultrapassar 2 segundos consecutivos.

### O que acontece quando a cota mensal da API é esgotada?
A CPFHub.io não bloqueia as requisições quando a cota é atingida. Cada consulta além das incluídas no plano é cobrada a R$0,15 automaticamente. O plano Gratuito inclui 50 consultas/mês e o plano Pro inclui 1.000 consultas por R$149/mês. Por isso, monitorar o consumo com alertas em 80% e 90% da cota é fundamental.

### Como garantir conformidade com a LGPD ao registrar logs de consulta?
Nunca registre o CPF completo em logs. Use mascaramento -- por exemplo, `123***00` -- e aplique controle de acesso ao sistema de logs. A [ANPD](https://www.gov.br/anpd) orienta que dados de identificação devem ser tratados com o princípio da necessidade, e logs com CPF completo raramente são necessários para diagnóstico.

### Quais ferramentas são recomendadas para dashboards de observabilidade da API de CPF?
Prometheus + Grafana é a combinação mais comum para exportar e visualizar métricas de integração. O [Grafana](https://grafana.com/docs/grafana/latest/) oferece painéis prontos para latência (histogramas), taxa de erro (alertas por threshold) e consumo acumulado (séries temporais). Para equipes menores, um relatório semanal gerado automaticamente em Python já resolve a maioria dos casos.

### Leia também

- [SLA de API de CPF: entenda os níveis de disponibilidade](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade)
- [Como implementar retry com backoff exponencial em consultas de API de CPF](https://cpfhub.io/blog/como-implementar-retry-backoff-exponencial-consultas-api-cpf)
- [Como monitorar e alertar falhas na integração com API de CPF usando Grafana](https://cpfhub.io/blog/como-monitorar-alertar-falhas-integracao-api-cpf-grafana)
- [Como implementar cache inteligente para respostas da API de CPF](https://cpfhub.io/blog/como-implementar-cache-inteligente-respostas-api-cpf)

---

## Conclusão

Monitorar o consumo e a saúde da integração com API de CPF não é um luxo -- é uma necessidade operacional. Sem observabilidade, problemas de latência, erros silenciosos e estouro de cota passam despercebidos até causarem impacto direto no negócio.

Com o wrapper monitorado, os alertas e os dashboards apresentados

Cadastre-se em [cpfhub.io](https://www.cpfhub.io/)