Building a Form Submission Agent: Processing and Responding to Web Form Entries

Why Form Submissions Need an AI Agent

Web forms are the front door for most businesses. Contact forms, demo requests, support inquiries, job applications — they all arrive as structured data that needs to be processed, validated, and responded to. The gap between a form submission and a meaningful response is where opportunities are won or lost.

Traditional form handlers send a generic confirmation email and dump the data into a spreadsheet. An AI agent can do dramatically better: classify the submission's intent, assess lead quality, generate a personalized response that addresses specific questions, route high-priority submissions to the right person immediately, and create CRM records with enriched context.

Form Submission Webhook Handler

Most form builders (Typeform, Gravity Forms, JotForm) support webhooks that fire when a form is submitted. Build a handler that accepts submissions from multiple forms.

flowchart LR
    CLIENT(["Client SDK"])
    GW["API Gateway<br/>auth plus rate limit"]
    APP["FastAPI app<br/>handlers and DI"]
    VAL["Pydantic validation"]
    SVC["Service layer<br/>business logic"]
    DB[(Database)]
    QUEUE[(Background queue)]
    OBS[(Tracing)]
    CLIENT --> GW --> APP --> VAL --> SVC
    SVC --> DB
    SVC --> QUEUE
    SVC --> OBS
    SVC --> CLIENT
    style GW fill:#4f46e5,stroke:#4338ca,color:#fff
    style APP fill:#f59e0b,stroke:#d97706,color:#1f2937
    style DB fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

import os
from fastapi import FastAPI, Request, BackgroundTasks
from pydantic import BaseModel, EmailStr
from openai import AsyncOpenAI
from datetime import datetime

app = FastAPI()
llm = AsyncOpenAI()

class FormSubmission(BaseModel):
    form_id: str
    submission_id: str
    submitted_at: datetime
    fields: dict[str, str]
    source_url: str | None = None
    ip_address: str | None = None
    utm_params: dict | None = None

@app.post("/forms/webhook/{form_id}")
async def receive_form_submission(
    form_id: str, request: Request, background_tasks: BackgroundTasks
):
    payload = await request.json()

    submission = FormSubmission(
        form_id=form_id,
        submission_id=payload.get("id", ""),
        submitted_at=datetime.utcnow(),
        fields=extract_fields(payload),
        source_url=payload.get("source_url"),
        utm_params=payload.get("utm"),
    )

    background_tasks.add_task(process_form_submission, submission)
    return {"status": "accepted", "submission_id": submission.submission_id}

def extract_fields(payload: dict) -> dict[str, str]:
    fields = {}
    for field in payload.get("fields", payload.get("answers", [])):
        label = field.get("label", field.get("field_name", "unknown"))
        value = field.get("value", field.get("answer", ""))
        if isinstance(value, dict):
            value = value.get("label", str(value))
        fields[label] = str(value)
    return fields

Intelligent Form Processing Pipeline

Route submissions through a pipeline that validates data, classifies intent, and triggers the appropriate workflow.

async def process_form_submission(submission: FormSubmission):
    validation = validate_submission(submission)
    if not validation["is_valid"]:
        await log_invalid_submission(submission, validation["errors"])
        return

    classification = await classify_submission(submission)

    response_text = await generate_response(submission, classification)
    email = submission.fields.get("email") or submission.fields.get("Email")
    if email:
        await send_personalized_response(email, response_text, submission)

    await route_submission(submission, classification)

    await create_crm_record(submission, classification)

def validate_submission(submission: FormSubmission) -> dict:
    errors = []
    fields = submission.fields

    email = fields.get("email") or fields.get("Email")
    if email and "@" not in email:
        errors.append("Invalid email format")

    message = fields.get("message") or fields.get("Message") or ""
    if len(message) < 10:
        errors.append("Message too short to process meaningfully")

    spam_indicators = ["buy now", "click here", "free offer", "act now"]
    message_lower = message.lower()
    if any(indicator in message_lower for indicator in spam_indicators):
        errors.append("Submission flagged as potential spam")

    return {"is_valid": len(errors) == 0, "errors": errors}

AI-Powered Submission Classification

Classify what the submitter wants and assess the quality of the lead.

FORM_CONFIGS = {
    "contact-form": {
        "name": "General Contact Form",
        "intents": ["sales_inquiry", "support_request",
                     "partnership", "press", "general"],
    },
    "demo-request": {
        "name": "Demo Request Form",
        "intents": ["enterprise_demo", "individual_demo", "partner_demo"],
    },
}

async def classify_submission(submission: FormSubmission) -> dict:
    form_config = FORM_CONFIGS.get(submission.form_id, {})
    fields_summary = "\n".join(
        f"  {k}: {v}" for k, v in submission.fields.items()
    )

    prompt = f"""Classify this form submission.

Form: {form_config.get('name', submission.form_id)}
Fields:
{fields_summary}
Source URL: {submission.source_url or 'Unknown'}
UTM Params: {submission.utm_params or 'None'}

Return a JSON object with:
- intent: the submitter's primary purpose
- lead_quality: score 1-10
- urgency: "immediate", "same_day", "next_day", "low"
- company_size_estimate: "enterprise", "mid_market", "small", "individual"
- key_interests: list of product/service areas mentioned
- summary: one sentence summary"""

    response = await llm.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    import json
    return json.loads(response.choices[0].message.content)

Personalized Response Generation

Generate a response that addresses the specific questions or needs expressed in the form.

async def generate_response(
    submission: FormSubmission, classification: dict
) -> str:
    fields_summary = "\n".join(
        f"  {k}: {v}" for k, v in submission.fields.items()
    )
    name = (
        submission.fields.get("name")
        or submission.fields.get("Name")
        or "there"
    )

    prompt = f"""Write a personalized email response to this form submission.

Submitter: {name}
Classification: {classification.get('intent')}
Their message:
{fields_summary}

Rules:
- Address their specific questions or needs
- If they asked for a demo, confirm timing and next steps
- If they have a support issue, acknowledge it and set expectations
- Include a specific call to action
- Keep it under 200 words
- Professional but warm tone
- Sign off as the team, not as an individual"""

    response = await llm.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

Routing and CRM Integration

Route high-value submissions immediately and create enriched CRM records.

import httpx

async def route_submission(submission: FormSubmission, classification: dict):
    urgency = classification.get("urgency", "low")
    lead_quality = classification.get("lead_quality", 1)

    if urgency == "immediate" or lead_quality >= 8:
        await send_slack_alert(
            channel="#hot-leads",
            message=(
                f"High-priority form submission!\n"
                f"Name: {submission.fields.get('name', 'Unknown')}\n"
                f"Intent: {classification.get('intent')}\n"
                f"Quality: {lead_quality}/10\n"
                f"Summary: {classification.get('summary')}"
            ),
        )

    if classification.get("intent") == "support_request":
        await create_support_ticket(submission, classification)

async def create_crm_record(submission: FormSubmission, classification: dict):
    crm_data = {
        "email": submission.fields.get("email") or submission.fields.get("Email"),
        "name": submission.fields.get("name") or submission.fields.get("Name"),
        "company": submission.fields.get("company") or submission.fields.get("Company"),
        "source": f"form:{submission.form_id}",
        "lead_score": classification.get("lead_quality", 1),
        "notes": classification.get("summary", ""),
        "utm_source": (submission.utm_params or {}).get("source"),
    }

    async with httpx.AsyncClient() as client:
        await client.post(
            f"{os.environ['CRM_API_BASE']}/contacts",
            headers={"Authorization": f"Bearer {os.environ['CRM_API_KEY']}"},
            json={"properties": crm_data},
        )

FAQ

How do I handle forms with file uploads?

Most form webhook providers send file URLs rather than the file content itself. Download the file from the provided URL, store it in your own object storage (S3, GCS), and pass the URL or extracted text content to the AI agent. Always validate file types and sizes before processing.

How fast should the response email arrive?

Under 5 minutes for sales and demo requests, under 15 minutes for general inquiries. Research shows that responding to leads within 5 minutes makes you 21 times more likely to qualify them compared to waiting 30 minutes. The AI agent makes sub-minute responses achievable.

How do I prevent duplicate CRM records from repeat submissions?

Check for existing contacts by email address before creating a new record. If a match exists, update the existing record with the new submission data and add a note. Use an upsert operation if your CRM API supports it, or implement check-then-create logic with a Redis lock to handle concurrent submissions.

---

#FormProcessing #AIAgents #LeadGeneration #FastAPI #CRMIntegration #AgenticAI #LearnAI #AIEngineering