AI Lead Firmographic Enrichment

id: ai-lead-firmographic-enrichment
namespace: company.team
description: |
  Automated inbound lead firmographic enrichment workflow using Perplexity AI web search. Extracts verified
  company size, industry, and funding details, stores records in PostgreSQL, and alerts sales on enterprise leads.

concurrency:
  limit: 1

inputs:
  - id: lead_email
    type: STRING
    displayName: Prospect Email
    defaults: "alex.taylor@stripe.com"
    description: "Email address of the inbound prospect."

  - id: company_name
    type: STRING
    displayName: Company Name
    defaults: "Stripe"
    description: "Company name provided by the prospect or extracted from the
      corporate email domain."

  - id: postgres_url
    type: STRING
    displayName: CRM Database JDBC URL
    defaults: "jdbc:postgresql://postgres.internal:5432/crm_db"
    description: "JDBC URL of the CRM database to store enriched lead profiles."

  - id: postgres_username
    type: STRING
    displayName: Database Username
    defaults: "crm_writer"
    description: "Database user with write permissions on the inbound_leads table."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#sales-leads"
    description: "Slack channel destination for notifying enterprise account executives."

triggers:
  - id: inbound_lead_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: "Executes in real-time upon receiving inbound form submission webhooks."
    key: "inbound-lead-enrichment"

tasks:
  - id: research_company_via_perplexity
    type: io.kestra.plugin.perplexity.ChatCompletion
    description: "Queries Perplexity online sonar model to perform live web research
      on target company firmographics."
    apiKey: "{{ secret('PERPLEXITY_API_KEY') }}"
    model: "sonar"
    messages:
      - role: user
        content: |
          Search the web and provide current firmographic data for company "{{ inputs.company_name }}".
          Output strictly valid JSON with the following keys:
          {
            "company_name": "{{ inputs.company_name }}",
            "industry": "...",
            "estimated_employees": 5000,
            "headquarters_location": "...",
            "funding_stage": "...",
            "is_enterprise": true
          }
          Set is_enterprise to true if estimated_employees >= 500, otherwise false.

  - id: parse_firmographic_data
    type: io.kestra.plugin.scripts.python.Script
    description: "Parses structured firmographic fields and classifies prospect lead tier."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      raw_perplexity.txt: "{{
        outputs.research_company_via_perplexity.choices[0].message.content }}"
    outputFiles:
      - lead_profile.json
    script: |
      import json
      import re
      from kestra import Kestra

      with open("raw_perplexity.txt", "r", encoding="utf-8") as f:
          raw = f.read().strip()

      clean_json = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.MULTILINE)
      clean_json = re.sub(r"\s*```$", "", clean_json, flags=re.MULTILINE).strip()

      try:
          profile = json.loads(clean_json)
      except Exception:
          profile = {
              "company_name": "{{ inputs.company_name }}",
              "industry": "Unknown",
              "estimated_employees": 100,
              "headquarters_location": "Unknown",
              "funding_stage": "Unknown",
              "is_enterprise": False
          }

      is_enterprise = bool(profile.get("is_enterprise", False))

      with open("lead_profile.json", "w", encoding="utf-8") as f:
          json.dump(profile, f, indent=2)

      Kestra.outputs({
          "is_enterprise": is_enterprise,
          "industry": str(profile.get("industry", "Unknown")),
          "employees": int(profile.get("estimated_employees", 0)),
          "hq": str(profile.get("headquarters_location", "Unknown")),
          "funding": str(profile.get("funding_stage", "Unknown"))
      })

  - id: evaluate_enterprise_tier
    type: io.kestra.plugin.core.flow.If
    description: "Branches execution to alert enterprise sales when company
      headcount exceeds 500."
    condition: "{{ outputs.parse_firmographic_data.vars.is_enterprise }}"
    then:
      - id: alert_enterprise_sales_lead
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches enterprise lead briefing card to Sales Slack channel."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*High-Value Enterprise Lead Alert*\n• *Prospect Email*: `{{ inputs.lead_email }}`\n• *Company*: `{{ inputs.company_name }}`\n• *Industry*: {{ outputs.parse_firmographic_data.vars.industry }}\n• *Estimated Headcount*: {{ outputs.parse_firmographic_data.vars.employees }}+ employees\n• *Headquarters*: {{ outputs.parse_firmographic_data.vars.hq }}\n• *Funding Stage*: {{ outputs.parse_firmographic_data.vars.funding }}\n• *Routing Action*: Enriched via Perplexity AI web search and assigned to Enterprise Account Executive queue."
          }

    else:
      - id: log_standard_lead_tier
        type: io.kestra.plugin.core.log.Log
        description: "Logs classification of mid-market or self-serve lead tier."
        message: "Inbound lead {{ inputs.lead_email }} enriched successfully. Classified
          as Mid-Market / Commercial account."

  - id: persist_lead_to_database
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: "Persists enriched firmographic dossier into PostgreSQL
      inbound_leads table."
    url: "{{ inputs.postgres_url }}"
    username: "{{ inputs.postgres_username }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      INSERT INTO inbound_leads (
        email, company_name, industry, employee_count, hq_location, is_enterprise, enriched_at
      ) VALUES (
        '{{ inputs.lead_email }}',
        '{{ inputs.company_name }}',
        '{{ outputs.parse_firmographic_data.vars.industry }}',
        {{ outputs.parse_firmographic_data.vars.employees }},
        '{{ outputs.parse_firmographic_data.vars.hq }}',
        {{ outputs.parse_firmographic_data.vars.is_enterprise }},
        CURRENT_TIMESTAMP
      )
      ON CONFLICT (email) DO UPDATE SET
        industry = EXCLUDED.industry,
        employee_count = EXCLUDED.employee_count,
        hq_location = EXCLUDED.hq_location,
        is_enterprise = EXCLUDED.is_enterprise,
        enriched_at = CURRENT_TIMESTAMP;

  - id: export_lead_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON lead manifest for marketing automation lineage."
    format: |
      {
        "pipeline": "ai-lead-firmographic-enrichment",
        "audited_at": "{{ execution.startDate }}",
        "company": "{{ inputs.company_name }}",
        "is_enterprise": {{ outputs.parse_firmographic_data.vars.is_enterprise }},
        "status": "ENRICHED"
      }