Lead-Anreicherung

version: "1.0"

name: lead-enricher
type: llm
model: connic/gpt-5.6-sol
description: "Enriches new signups with company research, ICP scoring, and lead storage"

system_prompt: |
  You enrich and qualify new signups. When a signup arrives, research the
  company and score it against your ideal customer profile (ICP).

  Enrichment workflow:
  1. Extract email, name, and company from the input.
  2. Check for an existing record using get_lead to avoid re-enriching duplicates.
  3. Use web_search to research the company: employee count, industry,
     funding stage, tech stack, and any recent relevant news.
  4. Use get_icp_criteria to load the scoring rubric.
  5. Score fit 0-100 based on:
     - Company size match (10-500 employees = high fit)
     - Industry alignment with your ICP
     - Budget signals (funding rounds, paid tooling, hiring for relevant roles)
     - Use case alignment with what your product solves
  6. Save the enriched lead with save_lead.
  7. Return a brief summary: company overview, score, status, and top 2-3 reasons.

  Score thresholds:
  - 70+: qualified (worth personal outreach)
  - 40-69: nurture (add to drip sequence)
  - <40: low (no immediate action)

  If the company cannot be found publicly, score 30 and note
  "insufficient public information".

temperature: 0.2
timeout: 60
max_iterations: 15

tools:
  - web_search
  - lead_tools.get_lead
  - lead_tools.get_icp_criteria
  - lead_tools.save_lead

"""Lead enrichment tools - domain wrappers over db_* and retrieval primitives.

Agents work with leads in business terms (save_lead, get_icp_criteria) rather
than constructing filter dicts or managing collection names and namespaces.
"""

from connic.tools import db_upsert, db_find, retrieval_query

LeadStatus = str  # "qualified" | "nurture" | "low"


async def save_lead(
    email: str,
    name: str,
    company: str,
    score: int,
    industry: str = "",
    company_size: str = "",
    funding_stage: str = "",
    notes: str = "",
) -> dict:
    """Save an enriched lead to the leads database.

    Args:
        email: Signup email address.
        name: Contact name.
        company: Company name.
        score: ICP fit score (0-100). 70+ = qualified, 40-69 = nurture, <40 = low.
        industry: Detected industry vertical.
        company_size: Estimated employee range (e.g. "50-200").
        funding_stage: Funding stage if found (e.g. "Series A", "Bootstrapped").
        notes: Research summary and score reasoning (2-4 sentences).

    Returns:
        The upsert result with "operation" ("inserted" or "updated") and
        "upserted_id". Keyed by email, so re-enriching the same signup updates
        the existing lead instead of creating a duplicate.
    """
    status: LeadStatus = (
        "qualified" if score >= 70 else "nurture" if score >= 40 else "low"
    )
    return await db_upsert(
        "leads",
        filter={"email": email},
        update={
            "name":          name,
            "company":       company,
            "score":         score,
            "status":        status,
            "industry":      industry,
            "company_size":  company_size,
            "funding_stage": funding_stage,
            "notes":         notes,
        },
        insert_only={"first_enriched_by": "lead-enricher"},
    )


async def get_lead(email: str) -> dict | None:
    """Look up an existing lead by email to avoid duplicate enrichment.

    Args:
        email: The signup email to look up.

    Returns:
        The lead document, or None if not found.
    """
    result = await db_find("leads", filter={"email": email}, limit=1)
    docs = result["documents"]
    return docs[0] if docs else None


async def get_icp_criteria(industry: str | None = None) -> dict:
    """Retrieve the ideal customer profile (ICP) scoring criteria.

    Reads from Retrieval (namespace "icp"). Seed it once with
    retrieval_store to describe your target customer, e.g.:

        retrieval_store(
            "Target: B2B SaaS companies with 20-500 employees, Series A or later,
             using modern cloud infra. High fit: fintech, HR tech, devtools.",
            namespace="icp",
            entry_id="icp-criteria",
        )

    Falls back to a generic rubric if nothing is stored yet.

    Args:
        industry: Optional industry to narrow the criteria query.

    Returns:
        Dict with source ("retrieval" or "default") and criteria list.
    """
    query = f"ideal customer profile scoring criteria {industry or ''}".strip()
    result = await retrieval_query(query, namespace="icp", max_results=3)
    items = result.get("results", [])
    if items:
        return {"source": "retrieval", "criteria": [r["content"] for r in items]}
    return {
        "source": "default",
        "criteria": [
            "B2B company with 10-500 employees",
            "Software, fintech, or e-commerce industry",
            "Clear automation or AI adoption signals",
        ],
    }

version: "1.0"

# The three lead_tools are mocked so no database or Retrieval data is touched.
# web_search is predefined and runs for real, so strict_mocks is intentionally
# omitted.
defaults:
  runs: 1
  timeout_s: 90
  mocks: lead_mocks

tests:
  - name: enriches_new_signup
    payload: '{"email":"jane@acme.com","name":"Jane Doe","company":"Acme Inc"}'
    expected_result: status == "completed"
    expected_tool_calls:
      - lead_tools.get_lead
      - lead_tools.get_icp_criteria
      - lead_tools.save_lead

  - name: looks_up_icp_rubric
    payload: '{"email":"sam@example.com","name":"Sam Lee","company":"Example Co"}'
    expected_result: status == "completed"
    expected_tool_calls:
      - lead_tools.get_icp_criteria