"""Phase B: AI-Powered Action Chain Suggestions.
Routes for suggesting and deploying action chains based on form context.
Wires chain_suggester.py and chain_templates.py to the API.
Endpoints:
POST /api/chains/suggest — Get chain suggestions for a site
GET /api/chains/templates — List available chain templates
POST /api/chains/deploy — Deploy a suggested chain as a real action
GET /api/chains/suggestions — List pending suggestions for a site
PUT /api/chains/suggestions/:id — Update a suggestion
DELETE /api/chains/suggestions/:id — Discard a suggestion
"""
from flask import Blueprint, jsonify, request, session
from app.models import (
TIERS,
check_chain_quota,
create_action,
get_actions_by_site,
get_site,
get_user,
get_user_chain_count,
)
from app.routes.auth import current_user, login_required
agent_build_bp = Blueprint("agent_build", __name__, url_prefix="/api/chains")
# ─── Chain Templates ────────────────────────────────────────────────────────────
@agent_build_bp.route("/templates", methods=["GET"])
@login_required
def list_templates():
"""List all available chain templates."""
from app.services.chain_templates import get_chain_templates_with_config
user = current_user()
# Filter by tier
tier = user.get("tier", "free")
tier_config = TIERS.get(tier, TIERS["free"])
templates = get_chain_templates_with_config()
return jsonify(
{
"templates": templates,
"tier": tier,
"tier_name": tier_config["name"],
}
), 200
@agent_build_bp.route("/templates/<template_id>", methods=["GET"])
@login_required
def get_template(template_id):
"""Get a specific chain template."""
from app.services.chain_templates import get_chain_template
template = get_chain_template(template_id)
if not template:
return jsonify({"error": "Template not found"}), 404
return jsonify({"template": template}), 200
# ─── Chain Suggestions ──────────────────────────────────────────────────────────
@agent_build_bp.route("/suggest", methods=["POST"])
@login_required
def suggest_chains():
"""Get AI-powered chain suggestions for a site.
Request:
{
"site_id": 123,
"prompt": "optional: describe what you want to happen after submission",
"use_llm": true # optional: force LLM vs template fallback
}
Response:
{
"suggestions": [...],
"source": "template" | "llm" | "template_fallback",
"site_id": 123,
"suggestion_id": "sc_abc123" # for later deploy/edit
}
"""
from app.services.chain_suggester import suggest_chains as template_suggest
from app.services.llm_client import suggest_chains_llm
data = request.get_json(force=True)
site_id = data.get("site_id")
prompt = data.get("prompt", "")
use_llm = data.get("use_llm", False)
if not site_id:
return jsonify({"error": "site_id is required"}), 400
user = current_user()
site = get_site(site_id)
if not site:
return jsonify({"error": "Site not found"}), 404
# Verify ownership
if user.get("id") != site.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
# Get form fields for context
import json as _json
field_config = site.get("field_config", "{}")
try:
fields = _json.loads(field_config) if isinstance(field_config, str) else field_config
except (_json.JSONDecodeError, TypeError):
fields = []
if fields is None:
fields = []
# Get form title/purpose from site
site_title = site.get("title", "")
if not prompt and site_title:
prompt = site_title
# Default prompt that triggers template matches
if not prompt or not prompt.strip():
prompt = "notify team on new submission"
# Generate suggestions
if use_llm:
result = suggest_chains_llm(
prompt=prompt or "general form",
fields=fields,
form_type=site.get("form_type", "form"),
)
suggestions = result.get("suggestions", [])
source = result.get("source", "template_fallback")
else:
suggestions = template_suggest(
prompt=prompt or "general form",
fields=fields,
form_type=site.get("form_type", "form"),
)
source = "template"
# Save suggestion to DB for later deploy/edit
suggestion_id = None
if suggestions:
suggestion_id = _save_suggestion(user["id"], site_id, suggestions, source)
return jsonify(
{
"suggestions": suggestions,
"source": source,
"site_id": site_id,
"suggestion_id": suggestion_id,
}
), 200
# ─── Suggestion CRUD ────────────────────────────────────────────────────────────
@agent_build_bp.route("/suggestions", methods=["GET"])
@login_required
def list_suggestions():
"""List pending chain suggestions for a site."""
from app.models_suggested_chains import get_suggestions_for_site
site_id = request.args.get("site_id")
if not site_id:
return jsonify({"error": "site_id query parameter is required"}), 400
user = current_user()
site = get_site(site_id)
if not site:
return jsonify({"error": "Site not found"}), 404
if user.get("id") != site.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
suggestions = get_suggestions_for_site(site_id)
return jsonify({"suggestions": suggestions}), 200
@agent_build_bp.route("/suggestions/<int:suggestion_id>", methods=["GET"])
@login_required
def get_suggestion(suggestion_id):
"""Get a specific chain suggestion."""
from app.models_suggested_chains import get_suggestion
suggestion = get_suggestion(suggestion_id)
if not suggestion:
return jsonify({"error": "Suggestion not found"}), 404
user = current_user()
if user.get("id") != suggestion.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
return jsonify({"suggestion": suggestion}), 200
@agent_build_bp.route("/suggestions/<int:suggestion_id>", methods=["PUT"])
@login_required
def update_suggestion(suggestion_id):
"""Update a chain suggestion (edit before deploy).
Request:
{
"chain_config": { ... full updated chain config ... }
}
"""
from app.models_suggested_chains import get_suggestion, update_suggestion
suggestion = get_suggestion(suggestion_id)
if not suggestion:
return jsonify({"error": "Suggestion not found"}), 404
user = current_user()
if user.get("id") != suggestion.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
data = request.get_json(force=True)
chain_config = data.get("chain_config")
if not chain_config:
return jsonify({"error": "chain_config is required"}), 400
update_suggestion(suggestion_id, chain_config=chain_config)
updated = get_suggestion(suggestion_id)
return jsonify({"suggestion": updated}), 200
@agent_build_bp.route("/suggestions/<int:suggestion_id>", methods=["DELETE"])
@login_required
def delete_suggestion(suggestion_id):
"""Discard a chain suggestion."""
from app.models_suggested_chains import delete_suggestion, get_suggestion
suggestion = get_suggestion(suggestion_id)
if not suggestion:
return jsonify({"error": "Suggestion not found"}), 404
user = current_user()
if user.get("id") != suggestion.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
delete_suggestion(suggestion_id)
return jsonify({"success": True}), 200
# ─── Deploy Chain ───────────────────────────────────────────────────────────────
@agent_build_bp.route("/deploy", methods=["POST"])
@login_required
def deploy_chain():
"""Deploy a suggested chain as a real action.
Request:
{
"site_id": 123,
"suggestion_id": "sc_abc123",
"chain_config": {
"name": "Lead Notification Workflow",
"steps": [...]
}
}
Response:
{
"success": true,
"action_id": 456,
"chain_steps": 3
}
"""
from app.services.chain_engine import validate_chain
data = request.get_json(force=True)
site_id = data.get("site_id")
suggestion_id = data.get("suggestion_id")
chain_config = data.get("chain_config")
if not site_id:
return jsonify({"error": "site_id is required"}), 400
if not chain_config and not suggestion_id:
return jsonify({"error": "chain_config or suggestion_id is required"}), 400
user = current_user()
site = get_site(site_id)
if not site:
return jsonify({"error": "Site not found"}), 404
if user.get("id") != site.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
# Load from suggestion if provided
if not chain_config and suggestion_id:
from app.models_suggested_chains import delete_suggestion, get_suggestion
suggestion = get_suggestion(suggestion_id)
if not suggestion:
return jsonify({"error": "Suggestion not found"}), 404
if user.get("id") != suggestion.get("user_id"):
return jsonify({"error": "Forbidden"}), 403
chain_config = suggestion.get("chain_config")
if isinstance(chain_config, str):
import json
try:
chain_config = json.loads(chain_config)
except json.JSONDecodeError:
return jsonify({"error": "Invalid chain config in suggestion"}), 400
# Check chain quota
allowed, current_count, max_chains = check_chain_quota(user["id"])
if not allowed:
tier_config = TIERS.get(user.get("tier", "free"), TIERS["free"])
if max_chains == 0:
return jsonify(
{
"error": "Action chains are not available on your current plan",
"upgrade_required": True,
"current_tier": user.get("tier", "free"),
}
), 403
return jsonify(
{
"error": f"Action chain limit reached ({current_count}/{max_chains})",
"upgrade_required": True,
"current_count": current_count,
"max_chains": max_chains,
"current_tier": user.get("tier", "free"),
}
), 403
# Validate chain config
validation = validate_chain(chain_config)
if not validation["valid"]:
return jsonify(
{
"error": "Invalid chain configuration",
"details": validation["errors"],
}
), 400
# Deploy!
action_id = create_action(
site_id=site_id,
action_type="chain",
config=chain_config,
trigger_event="submission",
enabled=True,
execution_order=len(get_actions_by_site(site_id)),
)
# Clean up suggestion
if suggestion_id:
from app.models_suggested_chains import delete_suggestion
delete_suggestion(suggestion_id)
chain_steps = len(chain_config.get("steps", []))
return jsonify(
{
"success": True,
"action_id": action_id,
"chain_steps": chain_steps,
}
), 201
# ─── Helpers ────────────────────────────────────────────────────────────────────
def _save_suggestion(user_id: int, site_id: int, suggestions: list, source: str) -> str:
"""Save a suggestion to the DB for later deploy/edit.
Returns suggestion_id or None if saving failed.
"""
try:
import json
from app.models_suggested_chains import create_suggestion
# Save first suggestion (or all as a batch)
chain_config = json.dumps(suggestions[0] if suggestions else {})
confidence = suggestions[0].get("confidence", 0.5) if suggestions else 0.5
category = suggestions[0].get("category", "general") if suggestions else "general"
return create_suggestion(
user_id=user_id,
site_id=site_id,
chain_config=chain_config,
source=source,
confidence=confidence,
category=category,
)
except Exception as e:
# Non-fatal — suggestions still work without DB persistence
import logging
logging.getLogger("agentforms.build").warning("Failed to save suggestion: %s", e)
return None