Monitoring Your Domain's Email Health with the Cleanbox API
If you manage email for a business — whether through Cleanbox aliases, relay, or both — you need visibility into what is happening. How many messages are being blocked? Are delivery failures spiking? Is the quarantine backlog growing? Are you approaching your weekly quota?
The Cleanbox API gives you all the data you need to answer these questions programmatically. In this guide, we build a Python monitoring script that runs daily, collects email health metrics, and posts a summary to Slack.
What we are monitoring
| Metric | API endpoint | Why it matters |
|---|---|---|
| Weekly usage vs. limit | GET /v1/team | Alerts before you hit your quota |
| Quarantine backlog | GET /v1/quarantine | Unreviewed quarantine = potential missed email |
| Delivery failures | GET /v1/messages?status=failed | Failed deliveries may indicate mailbox issues |
| Blocked messages | GET /v1/messages?status=denied | Spike in blocked = possible targeted spam campaign |
| Top spam senders | GET /v1/contacts?sort=emailCount | Identify senders that need to be blocked |
Step 1: Collect metrics
import requests
from datetime import datetime, timezone
API_BASE = "https://api.cleanbox.app/v1"
API_KEY = "your_api_key_here"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
def get(path, params=None):
r = requests.get(f"{API_BASE}{path}", headers=HEADERS, params=params)
r.raise_for_status()
return r.json()
def collect_metrics():
metrics = {}
# Team usage and limits
team = get("/team")
metrics["weekly_used"] = team["usage"]["weekly_used"]
metrics["weekly_limit"] = team["usage"]["weekly_limit"]
metrics["usage_pct"] = round(
(metrics["weekly_used"] / metrics["weekly_limit"]) * 100
) if metrics["weekly_limit"] > 0 else 0
# Alias and mailbox counts
metrics["aliases_used"] = team["limits"]["aliases"]["used"]
metrics["aliases_limit"] = team["limits"]["aliases"]["limit"]
metrics["mailboxes_used"] = team["limits"]["mailboxes"]["used"]
# Quarantine backlog
quarantine = get("/quarantine", {"per_page": 1})
metrics["quarantine_pending"] = quarantine.get("total_pages", 0) * 50 # rough estimate
# Recent delivery failures
failed = get("/messages", {"status": "failed", "per_page": 1})
metrics["failed_count"] = failed.get("total_pages", 0) * 25
# Recent blocked messages
denied = get("/messages", {"status": "denied", "per_page": 1})
metrics["denied_count"] = denied.get("total_pages", 0) * 25
# Top contacts by volume (potential spammers if unset state)
contacts = get("/contacts", {"sort": "emailCount", "dir": "desc", "per_page": 5})
metrics["top_senders"] = [
{"email": c["email"], "count": c["email_count"], "state": c["state"]}
for c in contacts["data"]
]
return metrics
Step 2: Analyze and generate alerts
def analyze(metrics):
alerts = []
# Quota warning at 80%
if metrics["usage_pct"] >= 80:
alerts.append(
f":warning: Weekly quota at {metrics['usage_pct']}% "
f"({metrics['weekly_used']}/{metrics['weekly_limit']})"
)
# Quarantine backlog
if metrics["quarantine_pending"] > 10:
alerts.append(
f":email: {metrics['quarantine_pending']}+ messages in quarantine awaiting review"
)
# Delivery failures
if metrics["failed_count"] > 5:
alerts.append(
f":x: {metrics['failed_count']}+ delivery failures detected - check mailbox connectivity"
)
# High-volume unset contacts (potential spam not yet blocked)
for sender in metrics["top_senders"]:
if sender["state"] == "unset" and sender["count"] > 50:
alerts.append(
f":mag: High-volume unset contact: {sender['email']} ({sender['count']} emails)"
)
return alerts
Step 3: Post to Slack
SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
def post_to_slack(metrics, alerts):
# Build summary
blocks = [
{
"type": "header",
"text": {"type": "plain_text", "text": ":mailbox_with_mail: Daily Email Health Report"}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Weekly Usage*
{metrics['weekly_used']}/{metrics['weekly_limit']} ({metrics['usage_pct']}%)"},
{"type": "mrkdwn", "text": f"*Aliases*
{metrics['aliases_used']}/{metrics['aliases_limit']}"},
{"type": "mrkdwn", "text": f"*Quarantine*
{metrics['quarantine_pending']}+ pending"},
{"type": "mrkdwn", "text": f"*Failed*
{metrics['failed_count']}+ messages"},
]
}
]
# Add alerts if any
if alerts:
alert_text = "
".join(alerts)
blocks.append({
"type": "section",
"text": {"type": "mrkdwn", "text": f"*Alerts*
{alert_text}"}
})
else:
blocks.append({
"type": "section",
"text": {"type": "mrkdwn", "text": ":white_check_mark: No alerts - everything looks healthy"}
})
requests.post(SLACK_WEBHOOK, json={"blocks": blocks})
# Run
metrics = collect_metrics()
alerts = analyze(metrics)
post_to_slack(metrics, alerts)
print(f"Report sent with {len(alerts)} alerts")
Schedule it
# Daily at 8 AM
0 8 * * * cd /path/to/script && python3 email_health.py >> /var/log/email_health.log 2>&1
Extending the monitor
- Trend tracking: Store daily metrics in a SQLite database and compare week-over-week to detect gradual changes
- Multi-team: If you manage multiple Cleanbox teams (e.g., as an MSP), run the script per team with different API keys
- Grafana integration: Push metrics to Prometheus/InfluxDB for dashboards and alerting
- Auto-remediation: If quarantine backlog exceeds a threshold, auto-accept messages from whitelisted contacts using
POST /v1/quarantine/accept