"""AgentForms application factory."""
import os
import signal
from datetime import datetime
from flask import Flask, current_app, jsonify, render_template, send_from_directory
from app.config import Config, validate
from app.middleware import init_middleware
from app.extensions import init_extensions
# ─── Blueprint imports ────────────────────────────────────────────────────────
from app.routes.actions import actions_bp
from app.routes.admin import admin_bp
from app.routes.agent_api import agent_api_bp
from app.routes.agent_build import agent_build_bp
from app.routes.agent_io import agent_io
from app.routes.agent_protocol import agent_protocol_bp
from app.routes.agents import agents_bp
from app.routes.api import api_bp
from app.routes.api_docs import api_docs_bp
from app.routes.auth import auth_bp
from app.routes.billing import billing_bp
from app.routes.chains import chains_bp
from app.routes.dashboard import dashboard_bp
from app.routes.documents import doc_bp
from app.routes.email import email_bp
from app.routes.integrations import export_bp, integrations_bp
from app.routes.invites import invite_bp
from app.routes.referral import referral_bp
from app.routes.site import site_bp
from app.routes.teams import team_bp
from app.routes.templates import templates_bp
from app.routes.tracking import tracking_bp
from app.routes.user_settings import user_settings_bp
from app.routes.user_sites import user_sites_bp
from app.routes.versions import versions_bp
from app.routes.webhooks import resend_bp
# ─── Graceful shutdown ────────────────────────────────────────────────────────
_shutting_down = False
def _graceful_shutdown(signum, frame):
"""On SIGTERM/SIGINT, drain in-flight requests before exiting."""
global _shutting_down
_shutting_down = True
from app.services.logging import log
log.info("shutdown", f"Received {signal.Signals(signum).name} — draining in-flight requests...")
signal.signal(signal.SIGTERM, _graceful_shutdown)
signal.signal(signal.SIGINT, _graceful_shutdown)
def create_app():
"""Flask application factory."""
# ─── Startup config validation (fail-fast) ────────────────────────────────
if not os.environ.get("TESTING", "").lower() == "true":
validate()
app = Flask(__name__)
# Apply configuration (SECRET_KEY, session, payload limits)
Config.apply(app)
# Register middleware (CSRF, CSP, security headers, error handlers, rate limiting)
init_middleware(app)
# ─── Static SEO/Compliance pages ──────────────────────────────────────────
@app.route("/robots.txt")
def robots_txt():
base = os.environ.get("APP_URL", "https://agentforms.io")
return (
f"User-agent: *\nAllow: /\nDisallow: /admin/\nDisallow: /settings/\nDisallow: /billing/\n\nSitemap: {base}/sitemap.xml\n",
200,
{"Content-Type": "text/plain"},
)
@app.route("/favicon.ico")
def favicon():
return send_from_directory(current_app.static_folder, "favicon.svg", mimetype="image/svg+xml")
@app.route("/.well-known/security.txt")
def security_txt():
base = os.environ.get("APP_URL", "https://agentforms.io")
return (
f"Contact: mailto:security@agentforms.io\nExpires: 2027-06-12T00:00:00.000Z\nEncryption: {base}/.well-known/security.txt.asc\nPreferred-Languages: en\nPolicy: {base}/security-policy\nCanonical: {base}/.well-known/security.txt\n",
200,
{"Content-Type": "text/plain"},
)
@app.route("/sitemap.xml")
def sitemap():
"""Dynamic sitemap including blog posts and all public routes."""
urls = []
base = os.environ.get("APP_URL", "https://agentforms.io")
static_pages = [
("/", 1.0, "daily"),
("/features", 0.8, "weekly"),
("/pricing", 0.9, "weekly"),
("/changelog", 0.6, "weekly"),
("/auth/register", 0.7, "monthly"),
("/auth/login", 0.7, "monthly"),
("/api/docs", 0.5, "monthly"),
("/wp-plugin", 0.4, "monthly"),
("/privacy", 0.3, "monthly"),
("/terms", 0.3, "monthly"),
]
now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S+00:00")
for path, priority, changefreq in static_pages:
urls.append(
f" <url>"
f"<loc>{base}{path}</loc>"
f"<lastmod>{now}</lastmod>"
f"<changefreq>{changefreq}</changefreq>"
f"<priority>{priority}</priority>"
f"</url>"
)
xml = '<?xml version="1.0" encoding="UTF-8"?>\n<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
xml += "\n".join(urls)
xml += "\n</urlset>"
return xml, 200, {"Content-Type": "application/xml"}
@app.route("/changelog")
def changelog_page():
"""Public changelog page."""
return render_template("changelog.html")
# ─── API Documentation ─────────────────────────────────────────────────────
@app.route("/api/docs")
def api_docs():
return render_template("api_docs.html")
# ─── Health endpoints ──────────────────────────────────────────────────────
@app.route("/health")
def health():
"""Liveness & readiness probe — checks DB, Redis, Stripe connectivity."""
from datetime import UTC, datetime
status = "healthy"
checks: dict = {}
# ── DB check ──
try:
from app.models import get_db
conn = get_db()
conn.execute("SELECT 1")
conn.close()
checks["db"] = "ok"
except Exception as e:
status = "degraded"
checks["db"] = f"error: {e}"
# ── Redis check ──
redis_url = os.environ.get("REDIS_URL")
if redis_url:
try:
import redis as redis_lib
r = redis_lib.from_url(redis_url)
r.ping()
checks["redis"] = "ok"
except Exception as e:
status = "degraded"
checks["redis"] = f"error: {e}"
else:
checks["redis"] = "not_configured"
# ── Stripe check ──
stripe_key = os.environ.get("STRIPE_SECRET_KEY")
if stripe_key and stripe_key not in ("", "sk_live_xxx", "sk_test_xxx"):
try:
import stripe
stripe.api_key = stripe_key
stripe.Balance.retrieve() # lightweight API call
checks["stripe"] = "ok"
except Exception as e:
status = "degraded"
checks["stripe"] = f"error: {e}"
else:
checks["stripe"] = "not_configured"
result = {
"status": status,
"db": checks.get("db", "unknown"),
"redis": checks.get("redis", "unknown"),
"stripe": checks.get("stripe", "unknown"),
"timestamp": datetime.now(UTC).isoformat(),
}
code = 200 if status == "healthy" else 503
return jsonify(result), code
@app.route("/health/ready")
def health_ready():
"""Readiness probe — checks DB + Redis connectivity."""
from app.models import get_db
errors = []
conn = None
try:
conn = get_db()
conn.execute("SELECT 1")
except Exception as e:
errors.append({"db": str(e)})
finally:
if conn:
conn.close()
redis_url = os.environ.get("REDIS_URL")
if redis_url:
try:
import redis as redis_lib
r = redis_lib.from_url(redis_url)
r.ping()
except Exception as e:
errors.append({"redis": str(e)})
if errors:
return jsonify({"status": "degraded", "service": "relay", "errors": errors}), 503
return jsonify({"status": "ok", "service": "relay"}), 200
@app.route("/health/redis")
def health_redis():
"""Check Redis connectivity — used by worker healthcheck."""
redis_url = os.environ.get("REDIS_URL")
if not redis_url:
return jsonify({"status": "unavailable", "reason": "REDIS_URL not set"}), 503
try:
import redis as redis_lib
r = redis_lib.from_url(redis_url)
r.ping()
return jsonify({"status": "ok", "service": "redis"}), 200
except Exception as e:
return jsonify({"status": "error", "service": "redis", "error": str(e)}), 503
@app.route("/health/worker")
def health_worker():
"""Check RQ worker health via Redis heartbeat."""
redis_url = os.environ.get("REDIS_URL")
if not redis_url:
return jsonify({"status": "unavailable", "reason": "REDIS_URL not set"}), 503
try:
import redis as redis_lib
import time as _time
import rq
from rq.registry import StartedJobRegistry
r = redis_lib.from_url(redis_url)
registry = StartedJobRegistry(connection=r)
active_jobs = len(registry.get_job_ids())
heartbeat_key = "agentforms:worker:heartbeat"
last_heartbeat = r.get(heartbeat_key)
if last_heartbeat:
last_ts = float(last_heartbeat)
age = _time.time() - last_ts
if age > 180: # stale if > 3 min
return jsonify(
{
"status": "degraded",
"service": "worker",
"active_jobs": active_jobs,
"heartbeat_age_seconds": round(age, 1),
}
), 200
return jsonify(
{
"status": "ok",
"service": "worker",
"active_jobs": active_jobs,
"heartbeat_age_seconds": round(age, 1),
}
), 200
return jsonify(
{
"status": "no_heartbeat",
"service": "worker",
"active_jobs": active_jobs,
}
), 200
except Exception as e:
return jsonify({"status": "error", "service": "worker", "error": str(e)}), 503
# Initialize extensions (DB, Redis, rate limiter, background workers)
init_extensions(app)
# ─── Register blueprints ──────────────────────────────────────────────────
# Site first (catches / before admin)
app.register_blueprint(templates_bp) # /api/v2/templates — template marketplace
app.register_blueprint(agent_api_bp) # /api/v2/ — agent API (must be before /api/)
app.register_blueprint(site_bp)
app.register_blueprint(api_bp)
app.register_blueprint(auth_bp)
app.register_blueprint(user_sites_bp)
app.register_blueprint(billing_bp)
app.register_blueprint(email_bp)
app.register_blueprint(admin_bp, url_prefix="/admin")
app.register_blueprint(user_settings_bp)
app.register_blueprint(team_bp)
app.register_blueprint(invite_bp)
app.register_blueprint(integrations_bp)
app.register_blueprint(export_bp)
app.register_blueprint(versions_bp)
app.register_blueprint(doc_bp)
app.register_blueprint(referral_bp)
app.register_blueprint(api_docs_bp) # /api/docs/
app.register_blueprint(resend_bp) # /resend/webhook
app.register_blueprint(tracking_bp) # /tracking/open, /tracking/click
app.register_blueprint(actions_bp) # /api/actions — form action CRUD
app.register_blueprint(agent_io) # /api/v2/agent — deploy forms, generate docs on demand
app.register_blueprint(agents_bp) # /api/v2/agents — Phase C agent registry
app.register_blueprint(agent_protocol_bp) # /api/v2/agent-protocol — Phase C protocol
app.register_blueprint(dashboard_bp)
app.register_blueprint(chains_bp)
app.register_blueprint(agent_build_bp) # /api/chains — Phase B: AI chain suggestions
return app
# Module-level app instance — used by tests and Gunicorn
app = create_app()
if __name__ == "__main__":
from app.services.logging import log
port = int(os.environ.get("PORT", 5060))
debug = os.environ.get("DEBUG", "false").lower() == "true"
log.info("relay", "Starting", port=port)
app.run(host="0.0.0.0", port=port, debug=debug)