"""Agent API v2 — programmatic form lifecycle for AI agents and integrations.
Authenticated via API key in the Authorization header:
Authorization: Bearer afk_live_...
All routes return JSON. Errors use standard HTTP status codes.
"""
import json
import logging
import os
import re
from functools import wraps
logger = logging.getLogger(__name__)
from flask import Blueprint, g, jsonify, request
# App-wide base URL (from env, fallback to production default)
APP_URL = os.environ.get("APP_URL", "https://agentforms.io")
from app.models import (
API_KEY_PREFIX,
TIERS,
_decrypt_submission_fields,
_get_site_owner_password_hash,
add_campaign_recipients,
add_site,
assign_recipients_to_variants,
cancel_reminder,
check_api_permission,
check_document_quota,
check_send_quota,
create_ab_test,
create_campaign,
create_custom_template,
# Phase 4
create_reminder,
delete_campaign,
delete_custom_template,
delete_site,
find_site_by_token,
generate_api_key,
get_ab_test_info,
get_campaign,
get_campaign_recipients,
get_custom_template,
get_custom_templates,
get_db,
get_email_settings,
get_reminders,
get_site_analytics,
get_site_analytics_summary,
get_site_version,
get_site_versions,
get_submission,
get_user,
get_user_sites,
get_user_tier,
get_version_limit,
list_api_keys,
list_campaigns,
parse_site_fields,
record_send,
revoke_api_key,
rollback_site_version,
schedule_reminders_for_campaign,
send_ab_winner_to_remaining,
site_device_stats,
site_geo_stats,
site_impression_stats,
update_campaign_status,
update_custom_template,
update_site_fields,
upsert_email_settings,
validate_api_key,
)
from app.services.campaign_emails import enqueue_campaign
from app.services.form_logic import build_config_response
def _get_user_password_hash(user_id):
"""Look up the password hash for a user (for submission decryption)."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
return dict(row)["password_hash"]
return None
agent_api_bp = Blueprint("agent_api", __name__, url_prefix="/api/v2")
# ─── Auth middleware ────────────────────────────────────────────────────────
def api_key_required(f):
"""Require a valid API key in the Authorization header.
Sets g.api_user_id and g.api_permissions on success.
"""
@wraps(f)
def decorated(*args, **kwargs):
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return jsonify({"error": "Missing or malformed Authorization header. Use 'Bearer afk_live_...'"}), 401
raw_key = auth_header[7:] # strip 'Bearer '
user_id, permissions = validate_api_key(raw_key)
if user_id is None:
return jsonify({"error": "Invalid API key"}), 401
g.api_user_id = user_id
g.api_permissions = permissions
return f(*args, **kwargs)
return decorated
def require_permission(action):
"""Check that the authenticated API key has the required permission.
Must be used after @api_key_required.
"""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
if not check_api_permission(g.api_permissions, action):
return jsonify({"error": f"Permission denied: '{action}' required"}), 403
return f(*args, **kwargs)
return decorated
return decorator
# ─── API Key verification (public — key IS the auth) ───────────────────────
@agent_api_bp.route("/verify", methods=["GET"])
def verify_key():
"""Verify an API key and return the associated tier.
Lightweight endpoint for relay/integrations to check subscription status.
No Bearer header needed — key is passed as query param.
Returns:
{"tier": "free"} or {"tier": "starter"}, etc.
401 if key is invalid.
"""
from app.models import get_user_tier
raw_key = request.args.get("key", "")
if not raw_key:
return jsonify({"error": "Missing 'key' query parameter"}), 400
user_id, _ = validate_api_key(raw_key)
if user_id is None:
return jsonify({"error": "Invalid API key"}), 401
tier_key, _ = get_user_tier(user_id)
return jsonify({"tier": tier_key, "key_prefix": raw_key[:12]})
# ─── API Key management ────────────────────────────────────────────────────
@agent_api_bp.route("/keys", methods=["GET"])
@api_key_required
def list_keys():
"""List all API keys for the authenticated user."""
keys = list_api_keys(g.api_user_id)
return jsonify({"keys": keys})
@agent_api_bp.route("/keys", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def create_key():
"""Create a new API key.
Request body:
name: Human-readable name (required)
permissions: Optional permissions dict
"""
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
if not name:
return jsonify({"error": "Key name is required"}), 400
# Validate permissions
permissions = data.get("permissions", None)
if permissions:
valid_perms = {"read_forms", "write_forms", "read_submissions", "delete_forms"}
for key in permissions:
if key not in valid_perms:
return jsonify({"error": f"Invalid permission: '{key}'. Valid: {list(valid_perms)}"}), 400
try:
result = generate_api_key(name, g.api_user_id, permissions)
except ValueError as e:
return jsonify({"error": str(e)}), 409
return jsonify(
{
"id": result["id"],
"key_prefix": result["key_prefix"],
"full_key": result["full_key"],
"name": result["name"],
"permissions": result["permissions"],
"created_at": None,
"warning": "Store the full_key securely. It will not be shown again.",
}
), 201
@agent_api_bp.route("/keys/<int:key_id>", methods=["DELETE"])
@api_key_required
@require_permission("delete_forms")
def revoke_key(key_id):
"""Revoke an API key."""
success = revoke_api_key(key_id, g.api_user_id)
if not success:
return jsonify({"error": "API key not found or already revoked"}), 404
return jsonify({"revoked": True, "id": key_id})
# ─── Form CRUD ──────────────────────────────────────────────────────────────
@agent_api_bp.route("/forms", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def list_forms():
"""List all forms for the authenticated user.
Query params:
limit: Max results (default 50, max 200)
offset: Page offset (default 0)
"""
limit = min(request.args.get("limit", 50, type=int), 200)
offset = max(request.args.get("offset", 0, type=int), 0)
sites = get_user_sites(g.api_user_id)
# Build response
forms = []
for site in sites:
field_config = parse_site_fields(site)
forms.append(
{
"id": site["id"],
"token": site["token"],
"name": site["name"],
"field_count": len(field_config) if field_config else 0,
"created_by": site.get("created_by", "manual"),
"created_at": site["created_at"],
}
)
total = len(forms)
paginated = forms[offset : offset + limit]
return jsonify(
{
"forms": paginated,
"total": total,
"limit": limit,
"offset": offset,
}
)
@agent_api_bp.route("/forms", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def create_form():
"""Create a new form.
Request body:
name: Form name (required)
fields: Array of field definitions
metadata: Optional JSON metadata
Field definition:
{
"name": "field_name",
"label": "Display Label",
"type": "text" | "email" | "tel" | "textarea" | "select",
"required": true,
"options": [...] // for select type
}
"""
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
if not name:
return jsonify({"error": "Form name is required"}), 400
fields = data.get("fields", [])
if not isinstance(fields, list):
return jsonify({"error": "'fields' must be an array"}), 400
if len(fields) == 0:
return jsonify({"error": "At least one field is required"}), 422
# Validate field definitions
valid_types = {"text", "email", "tel", "textarea", "select", "number", "date", "hidden"}
for i, field in enumerate(fields):
if not isinstance(field, dict):
return jsonify({"error": f"Field {i} must be an object"}), 400
if "name" not in field or "label" not in field or "type" not in field:
return jsonify({"error": f"Field {i} must have 'name', 'label', and 'type'"}), 400
if field["type"] not in valid_types:
return jsonify({"error": f"Field {i}: invalid type '{field['type']}'. Valid: {list(valid_types)}"}), 400
# Check tier site limit
user = get_user(g.api_user_id)
from app.models import get_user_site_count
current_count = get_user_site_count(g.api_user_id)
tier_config = TIERS.get(user["tier"], TIERS["free"])
max_sites = tier_config["max_sites"]
if max_sites > 0 and current_count >= max_sites:
return jsonify(
{
"error": "Site limit reached for your tier",
"current": current_count,
"max": max_sites,
}
), 403
metadata = data.get("metadata", None)
site = add_site(
name=name,
owner_email=user["email"],
user_id=g.api_user_id,
field_config=fields,
)
# Update created_by and metadata
conn = None
try:
conn = get_db()
updates = []
params = []
updates.append("created_by = ?")
params.append("agent")
if metadata:
updates.append("metadata = ?")
params.append(json.dumps(metadata) if isinstance(metadata, dict) else str(metadata))
params.append(site["id"])
conn.execute(f"UPDATE sites SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
# Refresh site data
site = conn.execute("SELECT * FROM sites WHERE id = ?", (site["id"],)).fetchone()
finally:
if conn:
conn.close()
site = dict(site)
return jsonify(
{
"id": site["id"],
"token": site["token"],
"name": site["name"],
"fields": fields,
"created_by": "agent",
"embed_url": f"{APP_URL}/embed.js",
"submit_url": f"{APP_URL}/api/submit?token={site['token']}",
"created_at": site["created_at"],
}
), 201
@agent_api_bp.route("/forms/<form_token>", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_form(form_token):
"""Get form details by token.
Returns full form definition including fields and settings.
"""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
field_config = parse_site_fields(site)
return jsonify(
{
"id": site["id"],
"token": site["token"],
"name": site["name"],
"fields": field_config,
"field_count": len(field_config) if field_config else 0,
"created_by": site.get("created_by", "manual"),
"metadata": json.loads(site["metadata"]) if site.get("metadata") else None,
"honeypot_enabled": bool(site.get("honeypot_enabled", 1)),
"spam_filter_enabled": bool(site.get("spam_filter_enabled", 1)),
"webhook_enabled": bool(site.get("webhook_enabled", 0)),
"rate_limit_enabled": bool(site.get("rate_limit_enabled", 1)),
"embed_url": f"{APP_URL}/embed.js",
"submit_url": f"{APP_URL}/api/submit?token={site['token']}",
"created_at": site["created_at"],
}
)
@agent_api_bp.route("/forms/<form_token>", methods=["DELETE"])
@api_key_required
@require_permission("delete_forms")
def delete_form(form_token):
"""Delete a form and all its submissions."""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
delete_site(site["id"])
return jsonify({"deleted": True, "id": site["id"], "token": form_token})
@agent_api_bp.route("/forms/<form_token>/fields", methods=["PUT"])
@api_key_required
@require_permission("write_forms")
def update_fields(form_token):
"""Update form fields.
Request body:
fields: Array of field definitions (replaces existing)
"""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
data = request.get_json(silent=True) or {}
fields = data.get("fields")
if fields is None:
return jsonify({"error": "'fields' is required"}), 400
if not isinstance(fields, list):
return jsonify({"error": "'fields' must be an array"}), 400
# Validate
valid_types = {"text", "email", "tel", "textarea", "select", "number", "date", "hidden"}
for i, field in enumerate(fields):
if not isinstance(field, dict):
return jsonify({"error": f"Field {i} must be an object"}), 400
if "name" not in field or "label" not in field or "type" not in field:
return jsonify({"error": f"Field {i} must have 'name', 'label', and 'type'"}), 400
if field["type"] not in valid_types:
return jsonify({"error": f"Field {i}: invalid type '{field['type']}'"}), 400
updated = update_site_fields(
site["id"], field_config=fields, changed_by="api_key", change_reason="Fields updated via agent API"
)
return jsonify(
{
"updated": True,
"fields": fields,
"field_count": len(fields),
}
)
# ─── Form config (public — no auth needed) ─────────────────────────────────
@agent_api_bp.route("/forms/<form_token>/config", methods=["GET"])
def get_form_config(form_token):
"""Get public form configuration.
Used by embed SDK and WordPress plugin to render forms.
No authentication required — only returns renderable data.
"""
site = find_site_by_token(form_token)
if not site:
return jsonify({"error": "Form not found"}), 404
field_config = parse_site_fields(site)
config = build_config_response(site, field_config)
config["submit_endpoint"] = f"/api/submit?token={site['token']}"
return jsonify(config)
# ─── Version history (authenticated) ──────────────────────────────────────
@agent_api_bp.route("/forms/<form_token>/versions", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def list_versions(form_token):
"""GET /api/v2/forms/{token}/versions — list version history for a form."""
site = find_site_by_token(form_token)
if not site:
return jsonify({"error": "Form not found"}), 404
if not check_api_permission(g.api_permissions, "read_forms"):
return jsonify({"error": "Access denied"}), 403
user_id = site.get("user_id")
if user_id:
tier_key, _ = get_user_tier(user_id)
limit = get_version_limit(tier_key)
else:
limit = None
versions = get_site_versions(site["id"], limit=limit)
return jsonify(
{
"site_id": site["id"],
"token": form_token,
"versions": versions,
"count": len(versions),
}
)
@agent_api_bp.route("/forms/<form_token>/versions/<int:version_id>", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_version(form_token, version_id):
"""GET /api/v2/forms/{token}/versions/{version} — get specific version."""
site = find_site_by_token(form_token)
if not site:
return jsonify({"error": "Form not found"}), 404
version = get_site_version(site["id"], version_id)
if not version:
return jsonify({"error": "Version not found"}), 404
return jsonify(
{
"site_id": site["id"],
"version": version,
}
)
@agent_api_bp.route("/forms/<form_token>/versions/<int:version_id>/rollback", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def rollback_version(form_token, version_id):
"""POST /api/v2/forms/{token}/versions/{version}/rollback — revert to version."""
site = find_site_by_token(form_token)
if not site:
return jsonify({"error": "Form not found"}), 404
data = request.get_json(silent=True) or {}
change_reason = data.get("reason", None)
target = get_site_version(site["id"], version_id)
if not target:
return jsonify({"error": "Version not found"}), 404
new_ver = rollback_site_version(
site["id"], version_id, changed_by=f"api_user:{g.api_user_id}", change_reason=change_reason
)
if new_ver is None:
return jsonify({"error": "Rollback failed"}), 500
return jsonify(
{
"rolled_back": True,
"new_version": new_ver,
"restored_from": version_id,
}
)
# ─── Submissions ────────────────────────────────────────────────────────────
@agent_api_bp.route("/forms/<form_token>/submissions", methods=["GET"])
@api_key_required
@require_permission("read_submissions")
def list_form_submissions(form_token):
"""List submissions for a form.
Query params:
limit: Max results (default 50, max 200)
offset: Page offset (default 0)
"""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
limit = min(request.args.get("limit", 50, type=int), 200)
offset = max(request.args.get("offset", 0, type=int), 0)
conn = None
try:
conn = get_db()
total_row = conn.execute(
"SELECT COUNT(*) as cnt FROM submissions WHERE site_id = ?",
(site["id"],),
).fetchone()
total = total_row["cnt"]
rows = conn.execute(
"""SELECT * FROM submissions WHERE site_id = ?
ORDER BY submitted_at DESC LIMIT ? OFFSET ?""",
(site["id"], limit, offset),
).fetchall()
finally:
if conn:
conn.close()
# Decrypt submissions using owner's password hash
pw_hash = _get_site_owner_password_hash(site["id"])
submissions = []
for row in rows:
d = dict(row)
# Decrypt PII + dynamic data (user-scoped encryption)
d = _decrypt_submission_fields(site["id"], d, pw_hash)
# Parse dynamic field data
if d.get("data"):
try:
d["dynamic_data"] = json.loads(d["data"])
except (json.JSONDecodeError, TypeError):
d["dynamic_data"] = {}
else:
d["dynamic_data"] = {}
submissions.append(d)
return jsonify(
{
"submissions": submissions,
"total": total,
"limit": limit,
"offset": offset,
}
)
@agent_api_bp.route("/forms/<form_token>/submissions/<int:sub_id>", methods=["GET"])
@api_key_required
@require_permission("read_submissions")
def get_form_submission(form_token, sub_id):
"""Get a single submission by ID."""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
submission = get_submission(sub_id, password_hash=_get_user_password_hash(g.api_user_id))
if not submission or submission["site_id"] != site["id"]:
return jsonify({"error": "Submission not found"}), 404
# Parse dynamic data
if submission.get("data"):
try:
submission["dynamic_data"] = json.loads(submission["data"])
except (json.JSONDecodeError, TypeError):
submission["dynamic_data"] = {}
else:
submission["dynamic_data"] = {}
return jsonify(submission)
# ── Analytics ──────────────────────────────────────────────────────────────
@agent_api_bp.route("/forms/<form_token>/analytics", methods=["GET"])
@api_key_required
@require_permission("read_submissions")
def form_analytics(form_token):
"""Get analytics for a form.
Tier-gated: Free gets basic only, Starter gets +geo/device, Pro+ gets everything.
Query params:
days: Days to include (default 30, max 90)
"""
site = find_site_by_token(form_token)
if not site or site.get("user_id") != g.api_user_id:
return jsonify({"error": "Form not found"}), 404
days = min(request.args.get("days", 30, type=int), 90)
# Resolve user tier and analytics level
tier_key, tier_config = get_user_tier(g.api_user_id)
analytics_level = tier_config["analytics"]
# Basic analytics (always available)
daily_data = get_site_analytics(site["id"], days)
summary = get_site_analytics_summary(site["id"])
result = {
"form_id": site["id"],
"form_token": site["token"],
"tier": tier_key,
"analytics_level": analytics_level,
"days": days,
"daily_submissions": [dict(d) for d in daily_data],
"summary": summary,
}
# Geo stats (Starter+)
if analytics_level in ["standard", "advanced"]:
result["geo"] = site_geo_stats(site["id"])
# Device stats (Starter+)
if analytics_level in ["standard", "advanced"]:
result["device"] = site_device_stats(site["id"])
# Impressions & conversion (Pro+)
if analytics_level == "advanced":
result["impressions"] = site_impression_stats(site["id"], days)
return jsonify(result)
# ─── Email Campaigns ────────────────────────────────────────────────────────
@agent_api_bp.route("/campaigns", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def list_campaigns_route():
"""List campaigns for the authenticated user.
Query params:
status: Filter by status (draft, scheduled, sending, sent, failed, cancelled)
limit: Max results (default 50, max 200)
offset: Page offset (default 0)
"""
status = request.args.get("status")
limit = min(request.args.get("limit", 50, type=int), 200)
offset = max(request.args.get("offset", 0, type=int), 0)
campaigns = list_campaigns(g.api_user_id, status=status, limit=limit, offset=offset)
return jsonify(
{
"campaigns": campaigns,
"total": len(campaigns),
"limit": limit,
"offset": offset,
}
)
@agent_api_bp.route("/campaigns", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def create_campaign_route():
"""Create a new email campaign.
Request body:
name: Campaign name (required)
subject: Email subject line (required)
body: Email body (HTML) (required)
from_name: Sender name (optional)
from_email: Sender email (optional)
template_id: Template ID (optional)
scheduled_at: ISO 8601 timestamp (optional)
recipients: Array of email addresses (optional)
"""
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
subject = data.get("subject", "").strip()
body = data.get("body", "").strip()
if not name:
return jsonify({"error": "Campaign name is required"}), 400
if not subject:
return jsonify({"error": "Email subject is required"}), 400
if not body:
return jsonify({"error": "Email body is required"}), 400
# Check tier email limit
user = get_user(g.api_user_id)
tier_config = TIERS.get(user["tier"], TIERS["free"])
max_campaigns = tier_config.get("max_campaigns", 0)
max_recipients = tier_config.get("max_recipients_per_campaign", 0)
if max_campaigns > 0:
existing = list_campaigns(g.api_user_id)
if len(existing) >= max_campaigns:
return jsonify(
{
"error": "Campaign limit reached for your tier",
"current": len(existing),
"max": max_campaigns,
}
), 403
recipients = data.get("recipients", [])
if max_recipients > 0 and len(recipients) > max_recipients:
return jsonify(
{
"error": f"Recipient limit exceeded. Max {max_recipients} per campaign.",
"count": len(recipients),
"max": max_recipients,
}
), 403
campaign = create_campaign(
user_id=g.api_user_id,
name=name,
subject=subject,
body=body,
from_name=data.get("from_name"),
from_email=data.get("from_email"),
template_id=data.get("template_id"),
scheduled_at=data.get("scheduled_at"),
)
# Add recipients
if recipients:
add_campaign_recipients(campaign["id"], g.api_user_id, recipients)
return jsonify(campaign), 201
@agent_api_bp.route("/campaigns/<int:campaign_id>", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_campaign_route(campaign_id):
"""Get campaign details including stats."""
campaign = get_campaign(campaign_id, g.api_user_id)
if not campaign:
return jsonify({"error": "Campaign not found"}), 404
return jsonify(campaign)
@agent_api_bp.route("/campaigns/<int:campaign_id>", methods=["DELETE"])
@api_key_required
@require_permission("delete_forms")
def delete_campaign_route(campaign_id):
"""Delete a campaign and its recipients."""
success = delete_campaign(campaign_id, g.api_user_id)
if not success:
return jsonify({"error": "Campaign not found"}), 404
return jsonify({"deleted": True, "id": campaign_id})
@agent_api_bp.route("/campaigns/<int:campaign_id>/recipients", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def list_recipients_route(campaign_id):
"""List recipients for a campaign.
Query params:
status: Filter by status (pending, sent, failed, bounced)
limit: Max results (default 100, max 500)
offset: Page offset (default 0)
"""
status = request.args.get("status")
limit = min(request.args.get("limit", 100, type=int), 500)
offset = max(request.args.get("offset", 0, type=int), 0)
recipients = get_campaign_recipients(campaign_id, g.api_user_id, status=status, limit=limit, offset=offset)
return jsonify(
{
"recipients": recipients,
"total": len(recipients),
"limit": limit,
"offset": offset,
}
)
@agent_api_bp.route("/campaigns/<int:campaign_id>/recipients", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def add_recipients_route(campaign_id):
"""Add recipients to a campaign.
Request body:
recipients: Array of email addresses (required)
"""
data = request.get_json(silent=True) or {}
recipients = data.get("recipients", [])
if not recipients or not isinstance(recipients, list):
return jsonify({"error": "recipients must be a non-empty array"}), 400
# Check tier recipient limit
user = get_user(g.api_user_id)
tier_config = TIERS.get(user["tier"], TIERS["free"])
max_recipients = tier_config.get("max_recipients_per_campaign", 0)
if max_recipients > 0:
existing = get_campaign_recipients(campaign_id, g.api_user_id, limit=99999)
if len(existing) + len(recipients) > max_recipients:
return jsonify(
{
"error": f"Recipient limit exceeded. Max {max_recipients} per campaign.",
"current": len(existing),
"adding": len(recipients),
"max": max_recipients,
}
), 403
result = add_campaign_recipients(campaign_id, g.api_user_id, recipients)
if "error" in result:
return jsonify(result), 404
return jsonify(result)
@agent_api_bp.route("/campaigns/<int:campaign_id>/send", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def send_campaign_route(campaign_id):
"""Send a campaign immediately.
Updates status to 'sending' and queues emails for delivery.
"""
campaign = get_campaign(campaign_id, g.api_user_id)
if not campaign:
return jsonify({"error": "Campaign not found"}), 404
if campaign["status"] not in ("draft", "scheduled"):
return jsonify({"error": f"Cannot send campaign with status '{campaign['status']}'"}), 400
# Update status
update_campaign_status(campaign_id, g.api_user_id, "sending")
# Queue for delivery
enqueue_campaign(campaign_id)
return jsonify(
{
"queued": True,
"campaign_id": campaign_id,
"status": "sending",
}
)
@agent_api_bp.route("/campaigns/<int:campaign_id>/cancel", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def cancel_campaign_route(campaign_id):
"""Cancel a scheduled or sending campaign."""
campaign = get_campaign(campaign_id, g.api_user_id)
if not campaign:
return jsonify({"error": "Campaign not found"}), 404
if campaign["status"] not in ("scheduled", "sending"):
return jsonify({"error": f"Cannot cancel campaign with status '{campaign['status']}'"}), 400
update_campaign_status(campaign_id, g.api_user_id, "cancelled")
return jsonify(
{
"cancelled": True,
"campaign_id": campaign_id,
"status": "cancelled",
}
)
@agent_api_bp.route("/campaigns/<int:campaign_id>/analytics")
@api_key_required
@require_permission("read_forms")
def campaign_analytics_route(campaign_id):
"""Get open/click/bounce analytics for a campaign."""
from app.models import get_campaign_analytics
campaign = get_campaign(campaign_id, g.api_user_id)
if not campaign:
return jsonify({"error": "Campaign not found"}), 404
analytics = get_campaign_analytics(campaign_id, g.api_user_id)
if analytics is None:
return jsonify({"error": "No analytics data"}), 404
return jsonify({"campaign_id": campaign_id, "analytics": analytics})
# ─── Email Settings ─────────────────────────────────────────────────────────
# ─── Direct email send (agent-facing) ──────────────────────────────────────────
@agent_api_bp.route("/email/send", methods=["POST"])
@api_key_required
def send_email_v2():
"""POST /api/v2/email/send — Send an email via the user's configured SMTP.
Available to paid tiers only. Free tier gets 403.
Request body:
to: Recipient email address (required)
subject: Email subject (required)
body: Plain-text or HTML body (required)
html: Optional HTML body (if provided, sent as HTML email)
from_addr: Optional override for SMTP_FROM
Response:
{\"success\": true, \"to\": \"...\", \"id\": \"...\"}
"""
from app.models import get_user, get_user_tier
from app.routes.email import NotificationService
user_id = g.api_user_id
# Tier gate: free tier cannot use direct email send
user = get_user(user_id)
if not user:
return jsonify({"error": "User not found"}), 404
tier_key, _ = get_user_tier(user_id)
tier_config = TIERS.get(tier_key, TIERS["free"])
max_sends = tier_config.get("max_api_email_sends", 0)
if max_sends <= 0:
return jsonify(
{
"error": "Direct email send requires a paid plan",
"tier": tier_key,
"upgrade_url": "/billing",
}
), 403
# Check rate limit
quota = check_send_quota(user_id)
if not quota.get("can_send", False):
return jsonify(
{
"error": "Send quota exceeded",
"quota": quota,
}
), 429
data = request.get_json(silent=True) or {}
to = data.get("to", "").strip()
subject = data.get("subject", "").strip()
body = data.get("body", "").strip()
html = data.get("html", None)
from_addr = data.get("from_addr", None)
if not to or not subject:
return jsonify({"error": "Missing required fields: to, subject"}), 400
if not body and not html:
return jsonify({"error": "At least one of 'body' or 'html' is required"}), 400
# Send the email
if html:
success = NotificationService.send_html(to, subject, html, body or None, from_addr)
else:
success = NotificationService.send(to, subject, body, from_addr)
if success:
# Record the send for rate limiting
record_send(user_id, count=1)
return jsonify({"success": True, "to": to})
else:
return jsonify({"error": "Failed to send email. Check SMTP configuration."}), 502
# ─── Email settings ──────────────────────────────────────────────────────────
@agent_api_bp.route("/email/settings", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_email_settings_route():
"""Get email sender settings."""
settings = get_email_settings(g.api_user_id)
return jsonify({"settings": settings})
@agent_api_bp.route("/email/settings", methods=["PUT"])
@api_key_required
@require_permission("write_templates")
def update_email_settings_route():
"""Update email sender settings.
Request body:
from_name: Sender display name
from_email: Sender email address
reply_to: Reply-to email address
template_id: Default template ID
sender_domain: Custom sender domain (e.g., mail.example.com)
default_template: Default template name
"""
data = request.get_json(silent=True) or {}
# Validate sender domain if provided
sender_domain = data.get("sender_domain")
if sender_domain is not None:
sender_domain = sender_domain.strip().lower()
if sender_domain and not re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", sender_domain):
return jsonify({"error": "Invalid domain format"}), 400
settings = upsert_email_settings(
user_id=g.api_user_id,
from_name=data.get("from_name"),
from_email=data.get("from_email"),
reply_to=data.get("reply_to"),
template_id=data.get("template_id"),
sender_domain=sender_domain,
domain_verified=data.get("domain_verified"),
default_template=data.get("default_template"),
)
return jsonify({"settings": settings})
# ─── Sender Domain Verification ───────────────────────────────────────────
@agent_api_bp.route("/email/domain/verify", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def verify_sender_domain_route():
"""Initiate sender domain verification.
Returns DNS records the user needs to add to verify ownership.
Request body:
domain: Domain to verify (required)
"""
data = request.get_json(silent=True) or {}
domain = data.get("domain", "").strip().lower()
if not domain:
return jsonify({"error": "Domain is required"}), 400
if not re.match(r"^[a-zA-Z0-9.-]+\.?[a-zA-Z]{2,}$", domain):
return jsonify({"error": "Invalid domain format"}), 400
# Generate verification token
import secrets
verification_token = secrets.token_hex(32)
# Store verification token (in practice, store in DB with expiry)
# For now, return the DNS records the user needs to add
dns_records = {
"txt": {
"name": f"_agentforms-verify.{domain}",
"value": f"agentforms-verification={verification_token}",
"ttl": 3600,
},
"dkim": {
"name": f"agentforms._domainkey.{domain}",
"value": f"v=DKIM1; k=rsa; p={generate_dkim_public_key_placeholder()}",
"ttl": 3600,
},
"spf": {
"name": domain,
"value": "include:agentforms.io",
"type": "add_to_existing",
"ttl": 3600,
},
"dmarc": {
"name": f"_dmarc.{domain}",
"value": "v=DMARC1; p=none; rua=mailto:dmarc@agentforms.io",
"ttl": 3600,
},
}
# Update email settings with domain and verification token
existing = get_email_settings(g.api_user_id)
upsert_email_settings(
user_id=g.api_user_id,
from_name=(existing or {}).get("from_name"),
from_email=(existing or {}).get("from_email"),
reply_to=(existing or {}).get("reply_to"),
template_id=(existing or {}).get("template_id"),
sender_domain=domain,
domain_verified=False,
default_template=(existing or {}).get("default_template"),
)
return jsonify(
{
"domain": domain,
"status": "pending",
"verification_token": verification_token,
"dns_records": dns_records,
"instructions": {
"title": "Verify Your Domain",
"steps": [
"Add the TXT record to your DNS to prove ownership",
"Add the DKIM record for email signing",
"Update your SPF record to include AgentForms",
"Add the DMARC record for policy enforcement",
"Click 'Verify' again once DNS propagation is complete",
],
"note": "DNS propagation can take up to 48 hours",
},
}
)
@agent_api_bp.route("/email/domain/check", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def check_sender_domain_route():
"""Check the verification status of the configured sender domain."""
settings = get_email_settings(g.api_user_id)
if not settings:
return jsonify(
{
"domain": None,
"status": "not_configured",
}
)
domain = settings.get("sender_domain")
if not domain:
return jsonify(
{
"domain": None,
"status": "not_configured",
}
)
return jsonify(
{
"domain": domain,
"status": "verified" if settings.get("domain_verified") else "pending",
"verified_at": settings.get("domain_verified_at") if settings.get("domain_verified") else None,
}
)
def generate_dkim_public_key_placeholder():
"""Generate a placeholder DKIM public key for display.
In production, this would generate a real key pair and store
the private key securely.
"""
# Placeholder — in production, generate actual RSA key pair
return "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQC..."
# ─── Email Templates ──────────────────────────────────────────────────────
@agent_api_bp.route("/templates", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def list_templates_route():
"""List available email templates.
Returns both built-in and user custom templates.
"""
from app.emails.renderer import _get_renderer
user_templates = get_custom_templates(g.api_user_id)
renderer = _get_renderer()
all_templates = renderer.list_templates()
# Separate built-in from custom
builtin = []
custom = []
for tpl in all_templates:
entry = {
"name": tpl["name"],
"type": tpl["type"],
}
if tpl["type"] == "custom":
# Check if this user owns it
user_tpl = next((t for t in user_templates if t["name"] == tpl["name"]), None)
if user_tpl:
entry.update(
{
"description": user_tpl.get("description"),
"created_at": user_tpl.get("created_at"),
"updated_at": user_tpl.get("updated_at"),
}
)
custom.append(entry)
else:
builtin.append(entry)
return jsonify(
{
"builtin_templates": builtin,
"custom_templates": custom,
}
)
@agent_api_bp.route("/templates", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def create_template_route():
"""Create a custom email template.
Request body:
name: Template name (required, unique per user)
content: HTML template content (required)
description: Template description (optional)
"""
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
content = data.get("content", "").strip()
if not name:
return jsonify({"error": "Template name is required"}), 400
if not content:
return jsonify({"error": "Template content is required"}), 400
# Validate name format
import re
if not re.match(r"^[a-zA-Z][a-zA-Z0-9_-]*$", name):
return jsonify(
{
"error": "Template name must start with a letter and contain only letters, numbers, hyphens, and underscores"
}
), 400
# Check tier limit (max_templates=0 means no custom templates allowed)
user = get_user(g.api_user_id)
tier_config = TIERS.get(user["tier"], TIERS["free"])
max_templates = tier_config.get("max_custom_templates", 0)
if max_templates >= 0:
existing = get_custom_templates(g.api_user_id)
if len(existing) >= max_templates:
return jsonify(
{
"error": "Custom template limit reached for your tier",
"current": len(existing),
"max": max_templates,
}
), 403
# Validate template syntax by attempting to render
from app.emails.renderer import _get_renderer
renderer = _get_renderer()
# Test render with empty context to catch syntax errors
try:
renderer.env.from_string(content)
except Exception as e:
return jsonify({"error": f"Invalid template syntax: {str(e)}"}), 400
template = create_custom_template(
user_id=g.api_user_id,
name=name,
content=content,
description=data.get("description"),
)
# Also save to disk for the renderer
renderer.save_custom_template(name, content)
return jsonify({"template": template}), 201
@agent_api_bp.route("/templates/<name>", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_template_route(name):
"""Get a custom template by name.
Returns template details and content preview.
"""
template = get_custom_template(g.api_user_id, name)
if not template:
return jsonify({"error": "Template not found"}), 404
# Don't return full content on GET — use separate endpoint
result = {
"name": template["name"],
"description": template.get("description"),
"created_at": template.get("created_at"),
"updated_at": template.get("updated_at"),
}
return jsonify({"template": result})
@agent_api_bp.route("/templates/<name>/content", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_template_content_route(name):
"""Get the raw content of a custom or built-in template."""
from app.emails.renderer import _get_renderer
renderer = _get_renderer()
# Check custom templates first
template = get_custom_template(g.api_user_id, name)
if template:
return jsonify(
{
"name": template["name"],
"type": "custom",
"content": template["content"],
"description": template.get("description"),
}
)
# Fall back to built-in
content = renderer.get_custom_template_content(name)
if content:
return jsonify(
{
"name": name,
"type": "builtin",
"content": content,
}
)
return jsonify({"error": "Template not found"}), 404
@agent_api_bp.route("/templates/<name>", methods=["PUT"])
@api_key_required
@require_permission("write_templates")
def update_template_route(name):
"""Update a custom email template.
Request body:
content: HTML template content (optional)
description: Template description (optional)
"""
data = request.get_json(silent=True) or {}
# Must have at least one field to update
if "content" not in data and "description" not in data:
return jsonify({"error": "No fields to update"}), 400
existing = get_custom_template(g.api_user_id, name)
if not existing:
return jsonify({"error": "Template not found"}), 404
content = data.get("content")
if content is not None:
# Validate syntax
from app.emails.renderer import _get_renderer
renderer = _get_renderer()
try:
renderer.env.from_string(content)
except Exception as e:
return jsonify({"error": f"Invalid template syntax: {str(e)}"}), 400
template = update_custom_template(
user_id=g.api_user_id,
name=name,
content=content,
description=data.get("description"),
)
if not template:
return jsonify({"error": "Template not found"}), 404
# Update disk copy if content changed
if content is not None:
from app.emails.renderer import _get_renderer
_get_renderer().save_custom_template(name, content)
return jsonify({"template": template})
@agent_api_bp.route("/templates/<name>", methods=["DELETE"])
@api_key_required
@require_permission("write_templates")
def delete_template_route(name):
"""Delete a custom email template."""
result = delete_custom_template(g.api_user_id, name)
if not result:
return jsonify({"error": "Template not found"}), 404
return jsonify({"message": f"Template '{name}' deleted"})
@agent_api_bp.route("/templates/render", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def render_template_route():
"""Render a template with the provided context (for preview/testing).
Request body:
template: Template name (required)
context: Dict of template variables (optional)
Returns rendered HTML.
"""
from app.emails.renderer import _get_renderer
data = request.get_json(silent=True) or {}
template_name = data.get("template", "").strip()
if not template_name:
return jsonify({"error": "Template name is required"}), 400
context = data.get("context", {})
# Merge in defaults
context.setdefault("base_url", os.environ.get("APP_URL", "https://agentforms.io"))
renderer = _get_renderer()
html = renderer.render(template_name, context)
return jsonify(
{
"template": template_name,
"html": html,
}
)
# --- Phase 4: Reminder Campaigns, A/B Testing, Rate Limiting ---
@agent_api_bp.route("/campaigns/<int:campaign_id>/reminders", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_campaign_reminders_route(campaign_id):
"""Get all reminders for a campaign."""
reminders = get_reminders(campaign_id, g.api_user_id)
return jsonify({"reminders": reminders})
@agent_api_bp.route("/campaigns/<int:campaign_id>/reminders", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def create_campaign_reminder_route(campaign_id):
"""Create a reminder for a campaign.
Request body:
delay_hours: Hours after campaign sent_at to send reminder
subject: Email subject (supports {first_name}, {email})
body: Email body HTML (supports {first_name}, {email})
template_id: Optional template ID to render body from
"""
data = request.get_json(silent=True) or {}
delay_hours = data.get("delay_hours")
subject = data.get("subject", "").strip()
body = data.get("body", "").strip()
template_id = data.get("template_id")
if not delay_hours or not isinstance(delay_hours, (int, float)) or delay_hours <= 0:
return jsonify({"error": "delay_hours must be a positive number"}), 400
if not subject:
return jsonify({"error": "subject is required"}), 400
if not body:
return jsonify({"error": "body is required"}), 400
reminder = create_reminder(campaign_id, g.api_user_id, delay_hours, subject, body, template_id)
if not reminder:
return jsonify({"error": "Campaign not found or access denied"}), 404
return jsonify({"reminder": reminder}), 201
@agent_api_bp.route("/campaigns/<int:campaign_id>/reminders/<int:reminder_id>", methods=["DELETE"])
@api_key_required
@require_permission("write_templates")
def cancel_campaign_reminder_route(campaign_id, reminder_id):
"""Cancel a scheduled reminder."""
success = cancel_reminder(reminder_id, g.api_user_id)
if not success:
return jsonify({"error": "Reminder not found or already sent"}), 404
return jsonify({"success": True})
@agent_api_bp.route("/email/rate-limit", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_rate_limit_route():
"""Get current rate limit quota for the user."""
quota = check_send_quota(g.api_user_id)
return jsonify(quota)
@agent_api_bp.route("/campaigns/<int:campaign_id>/ab-test", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def create_ab_test_route(campaign_id):
"""Create an A/B test for a campaign.
Request body:
variants: List of {label, subject, body, split_percent}
test_size: Percentage of recipients to include in test (default 20)
metric: Metric to measure winner (opens, clicks, responses)
declare_after_hours: Hours to wait before declaring winner (default 24)
"""
data = request.get_json(silent=True) or {}
variants = data.get("variants", [])
if not variants or len(variants) < 2:
return jsonify({"error": "At least 2 variants are required"}), 400
test_size = data.get("test_size", 20)
metric = data.get("metric", "opens")
declare_after_hours = data.get("declare_after_hours", 24)
if metric not in ("opens", "clicks", "responses"):
return jsonify({"error": "metric must be opens, clicks, or responses"}), 400
result = create_ab_test(campaign_id, g.api_user_id, variants, test_size, metric, declare_after_hours)
if not result:
return jsonify({"error": "Campaign not found or access denied"}), 404
return jsonify({"variants": result}), 201
@agent_api_bp.route("/campaigns/<int:campaign_id>/ab-test", methods=["GET"])
@api_key_required
@require_permission("read_forms")
def get_ab_test_route(campaign_id):
"""Get A/B test info for a campaign."""
info = get_ab_test_info(campaign_id, g.api_user_id)
if not info:
return jsonify({"error": "Campaign not found or no A/B test configured"}), 404
return jsonify(info)
@agent_api_bp.route("/campaigns/<int:campaign_id>/ab-test/assign", methods=["POST"])
@api_key_required
@require_permission("write_templates")
def assign_ab_variants_route(campaign_id):
"""Assign pending recipients to A/B test variants."""
count = assign_recipients_to_variants(campaign_id, g.api_user_id)
return jsonify({"assigned": count})
# ─── Document Generation (API-First) ──────────────────────────────────────────
@agent_api_bp.route("/documents/generate", methods=["POST"])
@api_key_required
@require_permission("write_forms")
def generate_document():
"""Generate a PDF document from JSON data via API key auth.
Auth: Bearer afk_live_* in Authorization header (no session required).
Request body (JSON):
document_type: "invoice" | "receipt" | "quote" | "statement" (required)
data: object with document field values (required)
For invoices: customer_name, customer_email, items[], total, currency, etc.
layout: "classic" | "modern" | "minimal" | "professional" (optional, default: "classic")
style: object with style overrides (optional)
e.g. {"primary_color": "#0066cc", "logo_url": "https://..."}
send_email: boolean (optional, default: false)
expires_in_days: integer (optional, default: 30, max: 365)
Response (JSON):
{
"document_id": "doc-abc123...",
"status": "completed",
"download_url": "https://agentforms.io/api/v2/documents/.../download",
"expires_at": "2026-08-06T12:00:00Z"
}
"""
from datetime import datetime, timedelta, timezone
from app.model_documents import (
generate_invoice_number,
store_document,
update_document_pdf_path,
)
from app.services.pdf_generator import generate_pdf, save_pdf
from app.services.document_builders import (
build_invoice_html_data,
build_receipt_html_data,
build_quote_html_data,
build_statement_html_data,
)
from app.services.documents import render_document_html
# ── Parse and validate request body ──────────────────────────────────────
payload = request.get_json(silent=True)
if not payload:
return jsonify({"error": "Request body must be valid JSON"}), 400
document_type = (payload.get("document_type") or "").strip().lower()
valid_types = {"invoice", "receipt", "quote", "statement"}
if document_type not in valid_types:
return jsonify(
{
"error": f"Invalid document_type '{document_type}'. Valid: {list(valid_types)}"
}
), 400
data = payload.get("data")
if not isinstance(data, dict):
return jsonify({"error": "'data' must be a JSON object"}), 400
layout = (payload.get("layout") or "classic").strip().lower()
valid_layouts = {"classic", "modern", "minimal", "professional"}
if layout not in valid_layouts:
return jsonify(
{
"error": f"Invalid layout '{layout}'. Valid: {list(valid_layouts)}"
}
), 400
style = payload.get("style", {}) or {}
if not isinstance(style, dict):
return jsonify({"error": "'style' must be a JSON object"}), 400
send_email = payload.get("send_email", False)
expires_in_days = min(payload.get("expires_in_days", 30), 365)
expires_in_days = max(expires_in_days, 1)
# ── Check document quota ────────────────────────────────────────────────
allowed, current_count, max_documents = check_document_quota(g.api_user_id)
if not allowed:
return jsonify(
{
"error": "Document generation limit reached for your tier",
"current_count": current_count,
"max_documents": max_documents,
}
), 429
# ── Build document data and HTML ────────────────────────────────────────
builders = {
"invoice": build_invoice_html_data,
"receipt": build_receipt_html_data,
"quote": build_quote_html_data,
"statement": build_statement_html_data,
}
try:
builder = builders[document_type]
html_data = builder(data, style, layout)
except Exception as e:
logger.error(
"document_builder_error",
extra={"user_id": g.api_user_id, "document_type": document_type, "error": str(e)},
)
return jsonify({"error": f"Failed to build document data: {str(e)}"}), 500
try:
html_content = render_document_html(html_data, style, layout)
except Exception as e:
logger.error(
"document_render_error",
extra={"user_id": g.api_user_id, "document_type": document_type, "error": str(e)},
)
return jsonify({"error": f"Failed to render document: {str(e)}"}), 500
# ── Generate PDF ────────────────────────────────────────────────────────
try:
pdf_bytes = generate_pdf(html_content, style)
except Exception as e:
logger.error(
"pdf_generation_error",
extra={"user_id": g.api_user_id, "document_type": document_type, "error": str(e)},
)
return jsonify({"error": f"Failed to generate PDF: {str(e)}"}), 500
# ── Store document record ───────────────────────────────────────────────
import uuid
document_id = f"doc-{uuid.uuid4().hex[:12]}"
expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days)
# Generate invoice number for invoices
invoice_number = None
if document_type == "invoice":
invoice_number = generate_invoice_number(g.api_user_id, data)
# Save PDF to disk
pdf_path = save_pdf(pdf_bytes, document_id)
# Store document record
store_document(
user_id=g.api_user_id,
document_id=document_id,
document_type=document_type,
data=data,
pdf_path=pdf_path,
expires_at=expires_at.isoformat(),
invoice_number=invoice_number,
customer_email=data.get("customer_email"),
)
# ── Optional: send email ────────────────────────────────────────────────
if send_email and data.get("customer_email"):
try:
from app.services.email_service import send_html
email_subject = f"{document_type.title()} {invoice_number or document_id}"
email_body = render_document_html(html_data, style, layout)
user = get_user(g.api_user_id)
send_html(
to=data["customer_email"],
subject=email_subject,
body=email_body,
from_addr=user.get("email", "noreply@agentforms.io"),
)
except Exception as e:
logger.warning(
"document_email_error",
extra={"user_id": g.api_user_id, "document_id": document_id, "error": str(e)},
)
# Don't fail the request if email fails
# ── Record API usage ────────────────────────────────────────────────────
try:
from app.models import record_api_usage
record_api_usage(g.api_user_id, "generate_document", document_type)
except Exception:
pass # Don't fail the request on usage tracking error
# ── Build response ──────────────────────────────────────────────────────
download_url = f"{APP_URL}/api/v2/documents/{document_id}/download"
return jsonify(
{
"document_id": document_id,
"status": "completed",
"download_url": download_url,
"expires_at": expires_at.isoformat(),
}
), 201