# ─── Agent I/O API ────────────────────────────────────────────────────────────
#
# Agent-facing endpoints. AI builds communication tools on demand via these endpoints.
# Agents use these to deploy forms for human input and generate documents for
# human output — without ever logging into a dashboard.
#
# Auth: Bearer <api_key> (uses existing API key table + usage tracking)
# ─── Imports ────────────────────────────────────────────────────────────────────
import json
import logging
import time
import uuid
from datetime import UTC, datetime, timedelta, timezone
from flask import Blueprint, current_app, jsonify, request
from app.db import get_db
from app.models import (
TIERS,
add_site,
add_submission,
check_rate_limit,
encrypt_value,
get_user,
get_user_site_count,
list_submissions,
record_api_usage,
validate_api_key,
)
logger = logging.getLogger("agent_io")
agent_io = Blueprint("agent_io", __name__, url_prefix="/api/v2/agent")
# ─── Constants ─────────────────────────────────────────────────────────────────
DEFAULT_FORM_TTL = 86400 * 30 # 30 days default
MAX_FORM_TTL = 86400 * 365 # 1 year max
MIN_FORM_TTL = 60 # 1 min min
# ─── Field Type Validation ────────────────────────────────────────────────────
VALID_FIELD_TYPES = [
"text",
"email",
"phone",
"number",
"textarea",
"select",
"checkbox",
"radio",
"date",
"time",
"datetime",
"url",
"file",
"signature",
"hidden",
"address",
"name",
"range",
"color",
"calculation",
]
# ─── Auth Helper ───────────────────────────────────────────────────────────────
def require_api_key():
"""Validate API key from Authorization header or X-Api-Key. Returns user dict or (error, status)."""
auth_header = request.headers.get("Authorization", "")
api_key = request.headers.get("X-Api-Key", "")
# Bearer token
if auth_header.startswith("Bearer "):
api_key = auth_header[7:]
if not api_key:
return None, ("API key required", 401)
# validate_api_key returns (user_id, permissions) or (None, None)
user_id, permissions = validate_api_key(api_key)
if user_id is None:
return None, ("Invalid API key", 401)
# Look up user info + API key ID
user = get_user(user_id)
if not user:
return None, ("User not found", 401)
# Look up API key ID for usage tracking
conn = get_db()
try:
key_row = conn.execute(
"SELECT id FROM api_keys WHERE user_id = ? AND key_prefix = ? AND is_active = 1",
(user_id, api_key[:12]),
).fetchone()
user["api_key_id"] = key_row["id"] if key_row else None
finally:
conn.close()
# Rate limit check
if not check_rate_limit(user_id, "api"):
return None, ("Rate limit exceeded", 429)
# Attach permissions for downstream use
user["api_permissions"] = permissions
return user, None
# ─── Deploy Form ──────────────────────────────────────────────────────────────
@agent_io.route("/deploy", methods=["POST"])
def deploy_form():
"""Deploy an ephemeral form.
Agent POSTs field configs + callback URL → gets back a hosted form URL.
Human fills form → structured data delivered to callback or retrievable via API.
Request body:
{
"fields": [...], // field config array (required)
"title": "Site Survey", // form title shown to human (required)
"callback_url": "...", // webhook URL for submissions (optional)
"branding": { // optional branding overlay
"name": "Company",
"logo": "https://...",
"primary_color": "#2563eb",
"tagline": "Professional plumbing services"
},
"ttl": 3600, // form expires in seconds (default 30 days)
"metadata": {}, // arbitrary agent state (returned in callbacks)
"require_auth": false, // gate submissions behind email
"success_message": "...", // override success message
"redirect_url": "...", // redirect after submission
}
Response:
{
"success": true,
"form_id": "site_123",
"form_url": "https://app.agentforms.io/f/abc123",
"embed_url": "https://app.agentforms.io/embed/abc123",
"token": "abc123",
"expires_at": "2025-06-22T...",
"submissions_url": "https://app.agentforms.io/api/v2/agent/submissions?token=abc123"
}
"""
user, err = require_api_key()
if err:
error_msg, status = err
return jsonify({"error": error_msg}), status
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "Invalid JSON body"}), 400
# ── Validate required fields ──
fields = data.get("fields")
title = data.get("title")
if not fields or not isinstance(fields, list) or len(fields) == 0:
return jsonify({"error": '"fields" is required. Array of field configs with at least one field.'}), 400
if not title:
return jsonify({"error": '"title" is required. Shown as the form heading.'}), 400
# ── Validate field configs ──
for i, field in enumerate(fields):
if "key" not in field:
return jsonify({"error": f"Field {i}: 'key' is required"}), 400
if "type" not in field:
field["type"] = "text" # default
if field["type"] not in VALID_FIELD_TYPES:
return jsonify(
{
"error": f"Field '{field['key']}': invalid type '{field['type']}'. Valid: {', '.join(VALID_FIELD_TYPES)}"
}
), 400
# ── Check tier site limit ──
user_id = user["id"]
current_count = get_user_site_count(user_id)
tier_config = TIERS.get(user.get("tier", "free"), TIERS["free"])
max_sites = tier_config.get("max_sites", 0)
if max_sites > 0 and current_count >= max_sites:
return jsonify(
{
"error": "Plan limit reached. Upgrade to deploy more forms.",
"current_plan": user.get("tier", "free"),
"current_sites": current_count,
"max_sites": max_sites,
}
), 403
# ── Build site config ──
token = uuid.uuid4().hex[:12]
branding = data.get("branding", {})
metadata = data.get("metadata", {})
# TTL handling
ttl = data.get("ttl", DEFAULT_FORM_TTL)
ttl = max(MIN_FORM_TTL, min(ttl, MAX_FORM_TTL))
expires_at = datetime.now(UTC) + timedelta(seconds=ttl)
# Build field_config for storage
field_config = {
"fields": fields,
"title": title,
"branding": branding,
"metadata": metadata,
"success_message": data.get("success_message", "Thank you! Your response has been recorded."),
"redirect_url": data.get("redirect_url", ""),
"require_auth": data.get("require_auth", False),
"expires_at": expires_at.isoformat(),
"deployed_by": "agent",
}
# ── Webhook URL (add_site encrypts internally) ──
callback_url = data.get("callback_url", "")
# ── Create the site ──
site = add_site(
name=title,
owner_email=user["email"],
smtp_from=None,
user_id=user["id"],
field_config=field_config, # add_site JSON-encodes this internally
webhook_url=callback_url, # add_site encrypts this internally
)
if not site or not site.get("id"):
return jsonify({"error": "Failed to create form. Please try again."}), 500
site_id = site["id"]
token = site["token"]
# ── Record usage ──
record_api_usage(user["id"], user["api_key_id"], "agent_deploy")
# ── Build response URLs ──
app_url = current_app.config.get("APP_URL", "https://agentforms.io")
form_url = f"{app_url}/f/{token}"
embed_url = f"{app_url}/embed/{token}"
logger.info(
"agent_deploy",
f"Deployed form '{title}' (token={token}, ttl={ttl}s) by user={user['id']}",
)
return jsonify(
{
"success": True,
"form_id": f"site_{site_id}",
"form_url": form_url,
"embed_url": embed_url,
"token": token,
"expires_at": expires_at.isoformat(),
"submissions_url": f"{app_url}/api/v2/agent/submissions?token={token}",
"ttl": ttl,
}
), 201
# ─── Generate Document ────────────────────────────────────────────────────────
@agent_io.route("/document", methods=["POST"])
def generate_document():
"""Generate a branded document from structured data.
Agent POSTs document type + data → gets back a PDF.
Optionally delivers to email or webhook.
Request body:
{
"document_type": "invoice", // invoice, quote, receipt, certificate, report
"data": { // structured document data (required)
"from": { "name": "...", "email": "...", "phone": "...", "address": "..." },
"to": { "name": "...", "email": "...", "address": "..." },
"line_items": [
{ "description": "...", "quantity": 1, "unit_price": 100, "amount": 100 }
],
"subtotal": 100,
"tax_rate": 0.08,
"tax": 8,
"total": 108,
"notes": "Payment due in 30 days",
"payment_terms": "Net 30",
},
"layout": "stripe", // stripe, hubspot, notion, classic, modern, etc.
"style": { // style overrides (optional)
"primary_color": "#2563eb",
"logo_url": "https://...",
},
"deliver": { // delivery method (optional)
"method": "email", // email, download, webhook
"to": "customer@email.com",
"subject": "Your Invoice",
"webhook_url": "https://...", // for webhook delivery
},
}
Response:
{
"success": true,
"document_id": "abc123",
"download_url": "https://app.agentforms.io/documents/abc123.pdf",
"size_bytes": 12345,
"delivered_to": "customer@email.com", // if email delivery
}
"""
user, err = require_api_key()
if err:
error_msg, status = err
return jsonify({"error": error_msg}), status
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "Invalid JSON body"}), 400
document_type = data.get("document_type", "invoice")
doc_data = data.get("data")
layout = data.get("layout", "modern")
style = data.get("style", {})
if not doc_data or not isinstance(doc_data, dict):
return jsonify({"error": '"data" is required. Structured document data dict.'}), 400
# ── Validate document type ──
valid_types = ["invoice", "quote", "receipt", "certificate", "report"]
if document_type not in valid_types:
return jsonify({"error": f"Invalid document_type '{document_type}'. Valid: {', '.join(valid_types)}"}), 400
try:
from app.services.documents import (
DEFAULT_STYLE,
generate_pdf,
render_document_html,
save_pdf,
)
# ── Merge styles ──
merged_style = {**DEFAULT_STYLE, **(style or {})}
# ── Generate document ID ──
document_id = str(uuid.uuid4())
# ── Enrich data with metadata ──
enriched_data = dict(doc_data)
enriched_data["document_id"] = document_id
enriched_data["document_type"] = document_type
enriched_data.setdefault("invoice_number", f"DOC-{datetime.now().strftime('%Y%m%d')}-{document_id[:6]}")
enriched_data.setdefault("issue_date", datetime.now(UTC).strftime("%B %d, %Y"))
# ── Render HTML → PDF ──
html_content = render_document_html(document_type, layout, enriched_data, merged_style)
pdf_bytes = generate_pdf(html_content, merged_style)
# ── Save PDF ──
pdf_path = save_pdf(pdf_bytes, document_id)
# ── Deliver ──
delivery_config = data.get("deliver", {})
delivery_method = delivery_config.get("method", "download")
delivered_to = None
if delivery_method == "email" and delivery_config.get("to"):
# Send PDF via email
try:
from app.utils.mailer import send_document_email
subject = delivery_config.get("subject", f"Your {document_type.title()}")
send_document_email(
to_address=delivery_config["to"],
subject=subject,
pdf_path=pdf_path,
document_type=document_type,
)
delivered_to = delivery_config["to"]
logger.info("agent_doc_delivered", f"Email sent to {delivered_to} for doc {document_id}")
except ImportError:
# Mailer not available — log and continue
logger.warning("agent_doc_email", f"Mailer not configured — skipping email delivery for {document_id}")
except Exception as e:
logger.error("agent_doc_email_fail", f"Email delivery failed: {e}")
elif delivery_method == "webhook" and delivery_config.get("webhook_url"):
# POST PDF info to webhook
import requests as req_lib
try:
webhook_payload = {
"event": "document_generated",
"document_id": document_id,
"document_type": document_type,
"download_url": f"{current_app.config.get('APP_URL', 'https://agentforms.io')}/documents/{document_id}.pdf",
"size_bytes": len(pdf_bytes),
"metadata": delivery_config.get("metadata", {}),
}
req_lib.post(
delivery_config["webhook_url"],
json=webhook_payload,
headers={"Content-Type": "application/json"},
timeout=10,
)
logger.info("agent_doc_webhook", f"Webhook delivered for doc {document_id}")
except ImportError:
logger.warning("agent_doc_webhook", "requests library not available for webhook delivery")
except Exception as e:
logger.error("agent_doc_webhook_fail", f"Webhook delivery failed: {e}")
# ── Record usage ──
record_api_usage(user["id"], user["api_key_id"], "agent_document")
app_url = current_app.config.get("APP_URL", "https://agentforms.io")
return jsonify(
{
"success": True,
"document_id": document_id,
"download_url": f"{app_url}/documents/{document_id}.pdf",
"size_bytes": len(pdf_bytes),
"layout": layout,
"delivered_to": delivered_to,
}
), 201
except Exception as e:
logger.error("agent_doc_gen_fail", f"Document generation failed: {e}")
return jsonify({"error": f"Document generation failed: {str(e)}"}), 500
# ─── Get Submissions ──────────────────────────────────────────────────────────
@agent_io.route("/submissions", methods=["GET"])
def get_submissions():
"""Retrieve submissions for a deployed form.
Query params:
token — form token (required)
limit — max results (default 50, max 200)
offset — pagination offset
Response:
{
"success": true,
"submissions": [
{
"id": 123,
"customer_name": "...",
"customer_email": "...",
"data": { ... },
"submitted_at": "2025-06-21T...",
}
],
"total": 42,
}
"""
user, err = require_api_key()
if err:
error_msg, status = err
return jsonify({"error": error_msg}), status
token = request.args.get("token")
if not token:
return jsonify({"error": "token query parameter required"}), 400
limit = min(int(request.args.get("limit", 50)), 200)
offset = int(request.args.get("offset", 0))
# Verify the user owns this form
conn = get_db()
try:
site = conn.execute("SELECT id, user_id FROM sites WHERE token = ?", (token,)).fetchone()
if not site:
return jsonify({"error": f"Form with token '{token}' not found"}), 404
if site["user_id"] != user["id"]:
return jsonify({"error": "Access denied. Form does not belong to this API key."}), 403
# Fetch submissions
rows = conn.execute(
"SELECT * FROM submissions WHERE site_id = ? ORDER BY submitted_at DESC LIMIT ? OFFSET ?",
(site["id"], limit, offset),
).fetchall()
total_row = conn.execute(
"SELECT COUNT(*) as cnt FROM submissions WHERE site_id = ?",
(site["id"],),
).fetchone()
total = total_row["cnt"] if total_row else 0
finally:
conn.close()
submissions = []
for row in rows:
sub = dict(row)
# Decrypt data field if needed
data_field = sub.get("data", "{}")
try:
sub["data"] = json.loads(data_field) if isinstance(data_field, str) else data_field
except (json.JSONDecodeError, TypeError):
sub["data"] = data_field
submissions.append(sub)
# Record usage
record_api_usage(user["id"], user["api_key_id"], "agent_submissions")
return jsonify(
{
"success": True,
"submissions": submissions,
"total": total,
"limit": limit,
"offset": offset,
}
), 200
# ─── Form Status ──────────────────────────────────────────────────────────────
@agent_io.route("/forms", methods=["GET"])
def list_forms():
"""List deployed forms for this API key.
Response:
{
"success": true,
"forms": [
{
"form_id": "site_123",
"token": "abc123",
"title": "...",
"form_url": "...",
"submissions_count": 5,
"expires_at": "2025-07-21T...",
"created_at": "2025-06-21T...",
}
],
}
"""
user, err = require_api_key()
if err:
error_msg, status = err
return jsonify({"error": error_msg}), status
conn = get_db()
try:
sites = conn.execute(
"SELECT s.*, COUNT(sub.id) as submissions_count "
"FROM sites s "
"LEFT JOIN submissions sub ON sub.site_id = s.id "
"WHERE s.user_id = ? "
"GROUP BY s.id "
"ORDER BY s.created_at DESC",
(user["id"],),
).fetchall()
finally:
conn.close()
app_url = current_app.config.get("APP_URL", "https://agentforms.io")
forms = []
for site in sites:
site_dict = dict(site)
field_config = site_dict.get("field_config", "{}")
try:
config = json.loads(field_config) if isinstance(field_config, str) else field_config
except (json.JSONDecodeError, TypeError):
config = {}
forms.append(
{
"form_id": f"site_{site_dict['id']}",
"token": site_dict["token"],
"title": config.get("title", site_dict.get("name", "Untitled")),
"form_url": f"{app_url}/f/{site_dict['token']}",
"embed_url": f"{app_url}/embed/{site_dict['token']}",
"submissions_count": site_dict.get("submissions_count", 0),
"expires_at": config.get("expires_at"),
"created_at": site_dict.get("created_at"),
}
)
return jsonify(
{
"success": True,
"forms": forms,
}
), 200
# ─── Delete Form ──────────────────────────────────────────────────────────────
@agent_io.route("/forms/<string:token>", methods=["DELETE"])
def delete_form(token):
"""Delete a deployed form and its submissions."""
user, err = require_api_key()
if err:
error_msg, status = err
return jsonify({"error": error_msg}), status
conn = get_db()
try:
site = conn.execute("SELECT id, user_id FROM sites WHERE token = ?", (token,)).fetchone()
if not site:
return jsonify({"error": f"Form '{token}' not found"}), 404
if site["user_id"] != user["id"]:
return jsonify({"error": "Access denied"}), 403
# Delete submissions first (FK constraint)
conn.execute("DELETE FROM submissions WHERE site_id = ?", (site["id"],))
conn.execute("DELETE FROM sites WHERE id = ?", (site["id"],))
conn.commit()
finally:
conn.close()
record_api_usage(user["id"], user["api_key_id"], "agent_delete")
logger.info("agent_delete", f"Deleted form {token} by user {user['id']}")
return jsonify({"success": True, "deleted": token}), 200