"""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/", 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/", 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/", 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//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//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//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//versions/", 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//versions//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//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//submissions/", 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//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/", 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/", 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//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//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//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//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//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/", 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//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/", 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/", 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//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//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//reminders/", 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//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//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//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