Recipe: lead enrichment & routing

What it does: a new lead comes in via the API; an HTTP connector enriches it, an LLM scores its fit 0–100, a branch routes hot / nurture / disqualified, and the owner is emailed.

When to use it: inbound lead routing, support-ticket triage, any “enrich → score → route” classification pipeline that runs unattended.

Flow

receive ─▶ [connector: enrich (HTTP)] ─▶ [connector: score (LLM)] ─▶ route
              │ onError → enrich_failed (continue)                     ├─ score ≥ 80 ─▶ route_hot ─▶ notify ─▶ done
                                                                       ├─ score ≥ 50 ─▶ route_nurture ─▶ notify ─▶ done
                                                                       └─ else ───────▶ route_disqualify ─▶ done

The workflow

// Lead enrichment & routing — enrich a new lead, score it, route it.
//
// A new lead arrives via the API; an HTTP connector enriches it, an LLM scores its
// fit 0–100, and a branch routes hot/warm/cold leads to different follow-ups, then
// emails the owner. Demonstrates HTTP + LLM connectors, branches, and a result reducer.
//
// Connections used: enrich (generic-http), llm, composio_gmail.
//
//   flowctl def create --file examples/workflows/lead-enrichment.js
//   flowctl run start --workflow LeadEnrichment \
//     --input '{"email":"jane@acme.com","company":"Acme"}' -q

function receive(ctx) { return { received: true }; }

function enrich_req(ctx) {
  return { query_params: { domain: ctx.vars.company } };
}
function enrich_resp(ctx, response) {
  return { enrichment: response.body, employees: (response.body && response.body.employees) || 0 };
}
function enrich_failed(ctx) {
  // Enrichment is best-effort — continue with what we have.
  return { enrichment: null, employees: 0 };
}

function score_req(ctx) {
  return {
    model: "anthropic/claude-3.5-sonnet",
    messages: [
      { role: "system", content: "Score B2B lead fit 0-100. Reply with only the number." },
      { role: "user", content: "Company: " + ctx.vars.company + " (" + ctx.vars.employees + " employees), email " + ctx.vars.email },
    ],
  };
}
function score_resp(ctx, response) {
  var n = parseInt(response.choices[0].message.content, 10);
  return { score: isNaN(n) ? 0 : n };
}

function route(ctx) { return {}; }

function route_hot(ctx) { return { queue: "sales", owner: "sales@example.com" }; }
function route_nurture(ctx) { return { queue: "nurture", owner: "marketing@example.com" }; }
function route_disqualify(ctx) { return { queue: "disqualified", owner: null }; }

function notify_req(ctx) {
  return { recipient_email: ctx.vars.owner, subject: "New " + ctx.vars.queue + " lead: " + ctx.vars.company, body: "Score " + ctx.vars.score };
}
function notify_resp(ctx, response) { return { notified: response.successful === true }; }

function done(ctx) { return { status: "routed" }; }

workflow({
  name: "LeadEnrichment",
  version: "1",
  description: "Enrich, score, and route an inbound lead.",
  initial: { email: null, company: null, score: 0, queue: null, owner: null },
  inputSchema: {
    type: "object",
    properties: { email: { type: "string" }, company: { type: "string" } },
    required: ["email", "company"],
  },
  triggers: [apiTrigger("new_lead")],
  steps: [
    step("receive", { start: true, next: "enrich", label: "Receive lead" }, receive),

    connector("enrich", {
      connection: "enrich",
      operation: "lookup_company",
      instance: "default",
      requestMapper: enrich_req,
      responseCombiner: enrich_resp,
      onError: "enrich_failed",
      next: "score",
    }),
    step("enrich_failed", { next: "score", label: "Enrichment skipped" }, enrich_failed),

    connector("score", {
      connection: "llm",
      operation: "chat",
      instance: "default",
      requestMapper: score_req,
      responseCombiner: score_resp,
      next: "route",
    }),

    step("route", {
      branches: [
        branch("score >= 80", "route_hot"),
        branch("score >= 50", "route_nurture"),
        branch("true", "route_disqualify"),
      ],
    }, route),

    step("route_hot", { next: "notify", label: "Hot lead" }, route_hot),
    step("route_nurture", { next: "notify", label: "Nurture" }, route_nurture),
    step("route_disqualify", { next: "done", label: "Disqualified" }, route_disqualify),

    connector("notify", {
      connection: "composio_gmail",
      operation: "send_email",
      instance: "my-gmail",
      requestMapper: notify_req,
      responseCombiner: notify_resp,
      next: "done",
    }),

    step("done", { next: "end", label: "Done" }, done),
  ],
  result: (vars) => ({ email: vars.email, company: vars.company, score: vars.score, queue: vars.queue }),
});

Run it

flowctl def create --file examples/workflows/lead-enrichment.js
flowctl run start --workflow LeadEnrichment \
  --input '{"email":"jane@acme.com","company":"Acme"}' -q
flowctl run status <exec> -o json     # see result.score / result.queue

Variations

• Unattended scoring uses the LLM connector. If a human is driving the run, swap it for an agent() step instead.
• More routes. Add branches (by region, by plan) and matching follow-up connectors.
• Write-back. Add a connector to update your CRM with the score and queue.