Agente de Lead Nurturing: da decisão ao follow-up personalizado

O problema do nurturing não é escrever email — é decidir o que fazer com cada lead, agora. Sequências estáticas tratam todo mundo igual. Um agente de nurturing lê o CRM, calcula score com regras explicáveis, escolhe o ângulo certo, gera copy em 3 passos, passa por guardrails e só envia depois de aprovação humana — ou escala para o SDR quando o lead está quente demais para automação.

Este não é um post sobre "use ChatGPT para emails". É um guia de implementação de produção: máquina de estados, HubSpot API com engagements, scoring híbrido (comportamental + ICP + decay temporal), approval gate com JWT scoped, webhooks de resposta e handoff estruturado para vendas.

Sumário executivo — Um agente de nurturing efetivo não substitui o SDR: ele filtra e qualifica leads frios até que um deles atinja o threshold de handoff. A arquitetura é uma máquina de estados com scheduler (APScheduler), fila persistente (SQLite), scoring híbrido e approval gate. O agente lê o CRM, calcula score, escolhe o ângulo de comunicação, gera copy em 3 passos, aplica guardrails e só envia após aprovação humana. Métricas-chave: taxa de qualificação, tempo entre touches, taxa de aprovação humana e custo por lead qualificado. O post inclui código Python real e integração com HubSpot.

Este guia cobre:

• 00 — Três leads, três decisões (walkthrough completo)
• 01 — A matemática do problema (por que SDRs não escalam)
• 02 — Arquitetura de produção (scheduler, fila, SQLite, webhooks)
• 03 — Máquina de estados do nurturing
• 04 — CRM Reader profundo (HubSpot v3 + engagements)
• 05 — Scoring híbrido com intent decay
• 06 — Copy em 3 passos (outline → draft → critique)
• 07 — Guardrails, rate limit e compliance
• 08 — Approval gate com JWT + Slack
• 09 — Envio, webhooks e classificador de respostas
• 10 — Handoff SDR, Hermes + MCP, avaliação e rollout

00 · Três Leads, Três Decisões Diferentes

Antes de qualquer código, veja o que o agente decide — não o que ele escreve. Três leads reais (nomes fictícios) que passaram pelo mesmo fluxo numa implementação típica de SaaS B2B:

Lead A — Marina, Head de Growth @ FinTech 80p

O agente não escreveu email. Gerou um brief estruturado para o SDR: histórico, score, páginas visitadas, sugestão de talk track ("mencionar case do setor financeiro, oferecer demo de 15 min").

Lead B — Ricardo, Analista de Marketing @ Agência 12p

Lead C — Patricia, CEO @ Indústria 200p

Insight central: o valor do agente está na decisão (HANDOFF, SEND, WAIT, STOP) — não no texto. Copy é consequência da decisão certa.

01 · O Problema Real: A Matemática que Ninguém Faz

O custo invisível do follow-up manual

Considere um time com 200 MQLs ativos e um SDR que gasta 15 minutos por lead para pesquisar CRM, decidir se faz sentido contactar e escrever um follow-up personalizado:

200 leads × 15 min = 3.000 min/mês = 50 horas/mês
50h × R$ 80/h (custo SDR) = R$ 4.000/mês só em follow-up de nurturing

Na prática, o SDR não consegue dar atenção aos 200. Prioriza os 20 que "parecem quentes" e ignora o resto — ou manda template genérico para escalar. Leads mornos com potencial (como Ricardo) recebem email ruim. Leads quentes (como Marina) esperam 3 dias porque o SDR está ocupado com leads frios.

Por que sequências estáticas não resolvem

Trigger: lead baixou ebook
→ Email 1 (dia 0): "Obrigado pelo download!"
→ Email 2 (dia 3): "Viu nosso case?"
→ Email 3 (dia 7): "Quer agendar uma call?"

O que um agente resolve (e o que não resolve)

Regra prática: se o email poderia ir para qualquer pessoa da lista sem mudar uma vírgula, é broadcast — não nurturing.

02 · Arquitetura de Produção

O diagrama abaixo é o que roda em produção — não um script de 50 linhas. Cada caixa é um módulo independente com responsabilidade única.

Componentes e responsabilidades

Estrutura de diretórios

nurture-agent/
├── main.py                 # orquestrador + scheduler
├── hubspot_client.py       # CRM reader com retry
├── scoring.py              # behavioral + ICP + decay
├── copywriter.py           # 3-pass generation
├── guardrails.py           # filtros e rate limit
├── approval.py             # JWT + Slack
├── webhooks.py             # handler de eventos
├── state_machine.py        # FSM
├── db.py                   # SQLite schema
├── config.yaml             # thresholds, ICP rules
└── nurture.db              # gerado em runtime

03 · Máquina de Estados

Cada lead vive em um estado persistido no SQLite. O scoring sugere ações; a FSM define o que é permitido. Sem FSM, você manda email de reativação para lead que já pediu demo.

# state_machine.py
from enum import Enum
from dataclasses import dataclass

class NurtureState(Enum):
    NEW = "new"
    WARMING = "warming"
    ENGAGED = "engaged"
    HOT = "hot"
    COOLING = "cooling"
    REACTIVATING = "reactivating"
    HANDOFF_SDR = "handoff_sdr"
    STOPPED = "stopped"
    CONVERTED = "converted"

TRANSITIONS: dict[NurtureState, set[NurtureState]] = {
    NurtureState.NEW: {NurtureState.WARMING},
    NurtureState.WARMING: {NurtureState.ENGAGED, NurtureState.COOLING, NurtureState.STOPPED},
    NurtureState.ENGAGED: {NurtureState.HOT, NurtureState.WARMING, NurtureState.COOLING},
    NurtureState.HOT: {NurtureState.HANDOFF_SDR, NurtureState.CONVERTED},
    NurtureState.COOLING: {NurtureState.REACTIVATING, NurtureState.STOPPED},
    NurtureState.REACTIVATING: {NurtureState.WARMING, NurtureState.STOPPED},
    NurtureState.HANDOFF_SDR: {NurtureState.CONVERTED, NurtureState.STOPPED},
    NurtureState.STOPPED: set(),
    NurtureState.CONVERTED: set(),
}

def transition(current: NurtureState, target: NurtureState, reason: str) -> dict:
    if target not in TRANSITIONS.get(current, set()):
        raise ValueError(f"Transição inválida: {current.value} → {target.value}")
    return {"from": current.value, "to": target.value, "reason": reason}

Regra crítica: quando state == HOT ou score_composto >= 75, o pipeline pula copywriter e gera brief para SDR. Marina (Lead A) cai aqui. Mandar email automático para lead quente é o erro mais caro do nurturing.

04 · CRM Reader Profundo (HubSpot v3)

Buscar só firstname e email não basta. O reader precisa de engagements (emails, reuniões, notas) e eventos de página.

Schema completo do LeadContext

# models.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional

class FunnelStage(Enum):
    LEAD = "lead"
    MQL = "mql"
    SQL = "sql"
    OPPORTUNITY = "opportunity"

class NurtureAction(Enum):
    SEND_DEMO_CTA = "send_demo_cta"
    SEND_EDUCATIONAL = "send_educational"
    HANDOFF_SDR = "handoff_sdr"
    WAIT = "wait"
    REACTIVATE = "reactivate"
    STOP = "stop"

@dataclass
class Engagement:
    type: str          # email, meeting, note, call
    timestamp: datetime
    subject: str
    body_preview: str
    opened: bool = False
    clicked: bool = False
    replied: bool = False

@dataclass
class PageEvent:
    url: str
    title: str
    timestamp: datetime
    visit_count: int = 1

@dataclass
class LeadContext:
    id: str
    name: str
    email: str
    company: str
    role: str
    industry: str
    employee_count: int
    stage: FunnelStage
    consent_status: str
    opted_out: bool
    last_interaction: Optional[datetime]
    engagements: list[Engagement] = field(default_factory=list)
    page_events: list[PageEvent] = field(default_factory=list)
    sdr_notes: str = ""
    hubspot_owner_id: str = ""
    days_in_stage: int = 0
    last_email_sent: Optional[datetime] = None
    emails_sent_count: int = 0

Cliente HubSpot com retry e rate limit

# hubspot_client.py
import os, time, requests
from datetime import datetime, timezone
from models import LeadContext, Engagement, PageEvent, FunnelStage

HUBSPOT_TOKEN = os.getenv("HUBSPOT_ACCESS_TOKEN")
BASE = "https://api.hubapi.com"

class HubSpotClient:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {HUBSPOT_TOKEN}"

    def _get(self, path: str, params: dict | None = None, retries: int = 3) -> dict:
        for attempt in range(retries):
            resp = self.session.get(f"{BASE}{path}", params=params, timeout=15)
            if resp.status_code == 429:
                time.sleep(int(resp.headers.get("Retry-After", 2)))
                continue
            resp.raise_for_status()
            return resp.json()
        raise RuntimeError(f"Rate limit persistente: {path}")

    def fetch_contact(self, contact_id: str) -> dict:
        props = [
            "firstname", "lastname", "email", "company", "jobtitle",
            "industry", "numemployees", "lifecyclestage", "hs_lead_status",
            "hs_email_optout", "hs_legal_basis", "notes_last_updated",
            "hubspot_owner_id", "hs_email_last_send_date", "num_contacted_notes",
        ]
        return self._get(f"/crm/v3/objects/contacts/{contact_id}",
                         {"properties": ",".join(props)})

    def fetch_engagements(self, contact_id: str, limit: int = 20) -> list[Engagement]:
        assoc = self._get(f"/crm/v4/objects/contacts/{contact_id}/associations/emails")
        results = []
        for item in assoc.get("results", [])[:limit]:
            eid = item["toObjectId"]
            email = self._get(f"/crm/v3/objects/emails/{eid}",
                              {"properties": "hs_email_subject,hs_email_text,hs_timestamp"})
            p = email.get("properties", {})
            ts = p.get("hs_timestamp")
            results.append(Engagement(
                type="email",
                timestamp=datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc),
                subject=p.get("hs_email_subject", ""),
                body_preview=(p.get("hs_email_text") or "")[:300],
            ))
        return results

    def build_lead_context(self, contact_id: str) -> LeadContext:
        data = self.fetch_contact(contact_id)
        p = data["properties"]
        stage_map = {
            "lead": FunnelStage.LEAD,
            "marketingqualifiedlead": FunnelStage.MQL,
            "salesqualifiedlead": FunnelStage.SQL,
            "opportunity": FunnelStage.OPPORTUNITY,
        }
        engagements = self.fetch_engagements(contact_id)
        last_ts = max((e.timestamp for e in engagements), default=None)

        return LeadContext(
            id=contact_id,
            name=f"{p.get('firstname','')} {p.get('lastname','')}".strip(),
            email=p.get("email", ""),
            company=p.get("company", ""),
            role=p.get("jobtitle", ""),
            industry=p.get("industry", ""),
            employee_count=int(p.get("numemployees") or 0),
            stage=stage_map.get(p.get("lifecyclestage", "lead"), FunnelStage.LEAD),
            consent_status=p.get("hs_legal_basis", "unknown"),
            opted_out=p.get("hs_email_optout") == "true",
            last_interaction=last_ts,
            engagements=engagements,
            sdr_notes=p.get("notes_last_updated", "") or "",
            hubspot_owner_id=p.get("hubspot_owner_id", ""),
            emails_sent_count=int(p.get("num_contacted_notes") or 0),
        )

Page events: se você usa HubSpot tracking, exponha visitas via propriedades customizadas (hs_analytics_last_url, hs_analytics_num_page_views) ou integre GA4 via webhook. Sem dado de página, remova o peso de pricing_visits do scoring — não invente.

05 · Scoring Híbrido com Intent Decay

Score composto = 40% comportamental + 30% ICP + 30% recência. ICP alto não salva lead inativo (Patricia, Lead C).

Scoring comportamental

# scoring.py
from datetime import datetime, timezone
from models import LeadContext, NurtureAction, NurtureState

BEHAVIOR_WEIGHTS = {
    "email_opened_7d": 15,
    "email_clicked_7d": 25,
    "pricing_visit_7d": 30,
    "demo_visit_7d": 20,
    "replied": 50,
    "meeting_booked": 60,
}

def decay_factor(days_ago: int) -> float:
    """Sinal perde metade do peso a cada 7 dias."""
    return 0.5 ** (days_ago / 7)

def score_behavioral(lead: LeadContext) -> tuple[int, list[str]]:
    score, reasons = 0, []
    now = datetime.now(timezone.utc)

    for eng in lead.engagements:
        days = (now - eng.timestamp).days
        factor = decay_factor(days)
        if eng.opened and days <= 7:
            pts = int(BEHAVIOR_WEIGHTS["email_opened_7d"] * factor)
            score += pts
            reasons.append(f"abriu email há {days}d (+{pts})")
        if eng.clicked and days <= 7:
            pts = int(BEHAVIOR_WEIGHTS["email_clicked_7d"] * factor)
            score += pts
            reasons.append(f"clicou há {days}d (+{pts})")
        if eng.replied:
            score += BEHAVIOR_WEIGHTS["replied"]
            reasons.append("respondeu email (+50)")

    for pe in lead.page_events:
        days = (now - pe.timestamp).days
        if days > 7:
            continue
        factor = decay_factor(days)
        if "/pricing" in pe.url:
            pts = int(BEHAVIOR_WEIGHTS["pricing_visit_7d"] * factor)
            score += pts
            reasons.append(f"visitou pricing há {days}d (+{pts})")

    if lead.last_interaction:
        silent = (now - lead.last_interaction).days
        if silent > 14:
            penalty = min(30, silent - 14)
            score -= penalty
            reasons.append(f"inatividade {silent}d (-{penalty})")

    return max(0, min(100, score)), reasons

Scoring ICP (determinístico)

ICP_TARGET = {
    "industries": {"financial services", "saas", "technology", "fintech"},
    "min_employees": 20,
    "max_employees": 500,
    "decision_roles": {"ceo", "cto", "vp", "head", "director", "founder"},
}

def score_icp(lead: LeadContext) -> tuple[int, list[str]]:
    score, reasons = 50, []  # baseline neutro

    if lead.industry.lower() in ICP_TARGET["industries"]:
        score += 20
        reasons.append(f"setor ICP: {lead.industry} (+20)")
    if ICP_TARGET["min_employees"] <= lead.employee_count <= ICP_TARGET["max_employees"]:
        score += 15
        reasons.append(f"porte {lead.employee_count}p (+15)")

    role_lower = lead.role.lower()
    if any(r in role_lower for r in ICP_TARGET["decision_roles"]):
        score += 15
        reasons.append(f"cargo decisor: {lead.role} (+15)")

    return max(0, min(100, score)), reasons

def composite_score(lead: LeadContext) -> dict:
    beh, beh_r = score_behavioral(lead)
    icp, icp_r = score_icp(lead)
    recency = 100 if lead.last_interaction and (datetime.now(timezone.utc) - lead.last_interaction).days <= 3 else (
        50 if lead.last_interaction and (datetime.now(timezone.utc) - lead.last_interaction).days <= 14 else 10
    )
    total = int(beh * 0.4 + icp * 0.3 + recency * 0.3)
    return {
        "composite": total,
        "behavioral": beh,
        "icp": icp,
        "recency": recency,
        "reasons": beh_r + icp_r,
    }

Decisão de ação

def decide_action(lead: LeadContext, scores: dict, state: NurtureState) -> NurtureAction:
    if lead.opted_out or lead.consent_status == "none":
        return NurtureAction.STOP
    if lead.stage in (FunnelStage.SQL, FunnelStage.OPPORTUNITY):
        return NurtureAction.HANDOFF_SDR
    if scores["composite"] >= 75 or state == NurtureState.HOT:
        return NurtureAction.HANDOFF_SDR
    if scores["composite"] >= 55:
        return NurtureAction.SEND_DEMO_CTA
    if scores["composite"] >= 35:
        return NurtureAction.SEND_EDUCATIONAL
    if scores["composite"] >= 15:
        return NurtureAction.WAIT
    return NurtureAction.REACTIVATE

Quando usar LLM no scoring (só casos ambíguos)

Se 35 <= composite <= 45 e o lead tem nota do SDR com texto livre, passe a nota para o LLM classificar intenção (interested, not_now, wrong_fit). Custo: ~200 tokens. Frequência: <10% dos leads. Nunca use LLM como score principal — auditoria e debugging ficam impossíveis.

06 · Copy em 3 Passos

Um único prompt gera copy medíocre. Pipeline de 3 passos separa estratégia, execução e crítica.

Passo 1 — Outline (estratégia)

OUTLINE_PROMPT = """Dado o contexto do lead, gere um outline JSON:
{"angle": "...", "hook": "...", "proof_point": "...", "cta_type": "...", "tone": "..."}
Regras: hook deve referenciar interação REAL do contexto. Sem urgência falsa."""

def generate_outline(lead: LeadContext, action: NurtureAction, scores: dict, llm) -> dict:
    context = f"""Lead: {lead.name} ({lead.role} @ {lead.company})
Ação: {action.value}
Score: {scores['composite']} — {', '.join(scores['reasons'][:5])}
Últimos engagements: {[e.subject for e in lead.engagements[:3]]}
Páginas: {[p.url for p in lead.page_events[:3]]}"""
    return llm.json(OUTLINE_PROMPT, context)

Passo 2 — Draft (execução)

DRAFT_PROMPT = """Escreva email de nurturing em português BR.
Máximo 130 palavras. Tom: {tone}. CTA: {cta_type}.
Use o hook: {hook}. Prova: {proof_point}.
Retorne JSON: {"subject": "...", "body": "..."}"""

def generate_draft(lead: LeadContext, outline: dict, llm) -> dict:
    prompt = DRAFT_PROMPT.format(**outline)
    return llm.json(prompt, f"Destinatário: {lead.name}")

Passo 3 — Self-critique (crítica)

CRITIQUE_PROMPT = """Avalie o email contra o contexto do lead.
Problemas a detectar: informação inventada, tom agressivo, CTA desalinhado, >150 palavras.
Se houver problemas, reescreva. Retorne JSON:
{"approved": bool, "issues": [...], "subject": "...", "body": "..."}"""

def critique_and_refine(lead: LeadContext, draft: dict, llm) -> dict:
    return llm.json(CRITIQUE_PROMPT, f"Contexto: {lead}\nDraft: {draft}")

Custo típico por lead: ~1.500 tokens (3 chamadas com Sonnet). Para 200 leads/dia: ~300k tokens ≈ R$ 15–25/dia. Compare com R$ 4.000/mês de tempo SDR.

07 · Guardrails, Rate Limit e SQLite

# db.py
import sqlite3
from datetime import datetime, timezone

SCHEMA = """
CREATE TABLE IF NOT EXISTS nurture_state (
    lead_id TEXT PRIMARY KEY,
    state TEXT NOT NULL,
    composite_score INTEGER,
    last_action TEXT,
    updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS audit_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    lead_id TEXT,
    event TEXT,
    detail TEXT,
    created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS email_sends (
    lead_id TEXT,
    sent_at TEXT NOT NULL,
    subject TEXT
);
"""

def init_db(path: str = "nurture.db"):
    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    return conn

def log_audit(conn, lead_id: str, event: str, detail: str):
    conn.execute(
        "INSERT INTO audit_log (lead_id, event, detail, created_at) VALUES (?,?,?,?)",
        (lead_id, event, detail, datetime.now(timezone.utc).isoformat())
    )
    conn.commit()

def rate_limit_ok(conn, lead_id: str, min_days: int = 5) -> bool:
    row = conn.execute(
        "SELECT sent_at FROM email_sends WHERE lead_id=? ORDER BY sent_at DESC LIMIT 1",
        (lead_id,)
    ).fetchone()
    if not row:
        return True
    last = datetime.fromisoformat(row[0])
    return (datetime.now(timezone.utc) - last).days >= min_days

# guardrails.py
import re

BANNED_PHRASES = [
    r"última chance", r"últimas vagas", r"só hoje", r"imperdível",
    r"100% garantido", r"sem risco", r"clique agora ou",
]
PII_PATTERNS = [r"\b\d{3}\.\d{3}\.\d{3}-\d{2}\b", r"\b\d{11}\b"]

def check_banned(text: str) -> list[str]:
    return [p for p in BANNED_PHRASES if re.search(p, text, re.I)]

def check_hallucination(body: str, lead: LeadContext) -> list[str]:
    issues = []
    if lead.company and lead.company.lower() not in body.lower():
        if "@" not in body:  # não mencionou empresa nem email
            issues.append("copy não referencia empresa do lead")
    return issues

def validate_email(body: str, subject: str, lead: LeadContext) -> tuple[bool, list[str]]:
    issues = check_banned(body) + check_banned(subject) + check_hallucination(body, lead)
    if len(body.split()) > 150:
        issues.append(f"corpo tem {len(body.split())} palavras (max 150)")
    return len(issues) == 0, issues

08 · Approval Gate com JWT + Slack

Padrão idêntico ao post de identidade e permissões: agente propõe, humano aprova, token scoped autoriza envio.

# approval.py
import jwt, uuid, os, requests
from datetime import datetime, timedelta, timezone

SECRET = os.getenv("NURTURE_JWT_SECRET")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")

def issue_send_token(agent_id: str, draft_id: str, lead_id: str, reviewer: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": agent_id,
        "draft_id": draft_id,
        "lead_id": lead_id,
        "approved_by": reviewer,
        "scope": "send:email",
        "jti": str(uuid.uuid4()),
        "iat": now,
        "exp": now + timedelta(minutes=15),
    }
    return jwt.encode(payload, SECRET, algorithm="HS256")

def notify_slack(draft: dict, lead: LeadContext, scores: dict):
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": f"Nurture: {lead.name}"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": (
            f"*Score:* {scores['composite']} | *Ação:* {draft['action']}\n"
            f"*Motivos:* {', '.join(scores['reasons'][:3])}\n"
            f"*Assunto:* {draft['subject']}\n```{draft['body'][:500]}```"
        )}},
        {"type": "actions", "elements": [
            {"type": "button", "text": {"type": "plain_text", "text": "Aprovar"},
             "style": "primary", "value": draft["draft_id"], "action_id": "approve"},
            {"type": "button", "text": {"type": "plain_text", "text": "Rejeitar"},
             "value": draft["draft_id"], "action_id": "reject"},
        ]},
    ]
    requests.post(SLACK_WEBHOOK, json={"blocks": blocks}, timeout=10)

09 · Envio, Webhooks e Classificador de Respostas

Envio via HubSpot Single Send

def send_email(token: str, draft: dict, lead: LeadContext, conn):
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    if payload.get("scope") != "send:email":
        raise PermissionError("Token sem escopo send:email")

    # HubSpot Marketing Email API ou transactional
    requests.post(
        f"{BASE}/marketing/v3/transactional/single-email/send",
        headers={"Authorization": f"Bearer {HUBSPOT_TOKEN}"},
        json={
            "emailId": os.getenv("HUBSPOT_TEMPLATE_ID"),
            "message": {"to": lead.email, "subject": draft["subject"], "body": draft["body"]},
        },
        timeout=15,
    )
    conn.execute("INSERT INTO email_sends VALUES (?,?,?)",
                 (lead.id, datetime.now(timezone.utc).isoformat(), draft["subject"]))
    log_audit(conn, lead.id, "sent", f"draft={payload['draft_id']}")
    conn.commit()

Webhook handler (resposta do lead)

# webhooks.py
REPLY_CLASSES = ["positive", "objection", "not_now", "opt_out", "ooo", "unrelated"]

def classify_reply(body: str, llm) -> str:
    return llm.classify(
        f"Classifique a resposta do lead em: {REPLY_CLASSES}. Texto: {body}"
    )

def handle_reply(lead_id: str, body: str, conn, llm):
    label = classify_reply(body, llm)
    log_audit(conn, lead_id, "reply", label)

    if label == "opt_out":
        transition_state(conn, lead_id, NurtureState.STOPPED, "opt-out via reply")
    elif label == "positive":
        transition_state(conn, lead_id, NurtureState.HANDOFF_SDR, "resposta positiva")
    elif label == "not_now":
        # agenda recontato em 30 dias
        schedule_followup(conn, lead_id, days=30)

10 · Handoff SDR

Quando action == HANDOFF_SDR, o agente gera brief JSON — não email:

def build_sdr_brief(lead: LeadContext, scores: dict) -> dict:
    return {
        "lead_id": lead.id,
        "name": lead.name,
        "company": lead.company,
        "role": lead.role,
        "score": scores["composite"],
        "signals": scores["reasons"],
        "recent_activity": [
            {"type": e.type, "subject": e.subject, "when": e.timestamp.isoformat()}
            for e in lead.engagements[:5]
        ],
        "pages_visited": [p.url for p in lead.page_events[:5]],
        "talk_track": (
            f"{lead.name} é {lead.role} na {lead.company}. "
            f"Sinais: {', '.join(scores['reasons'][:3])}. "
            f"Sugestão: demo de 15 min focada em {lead.industry}."
        ),
        "urgency": "high" if scores["composite"] >= 75 else "medium",
        "owner_id": lead.hubspot_owner_id,
    }

Poste no Slack #vendas ou crie task no HubSpot via API. SLA interno: SDR contacta em <4 horas para leads urgency: high.

11 · Hermes Agent + MCP

Alternativa ao script Python: orquestrador via Hermes Agent com CRM via MCP.

# ~/.config/hermes/config.yaml
agents:
  nurture-agent:
    model: claude-sonnet-4-20250514
    system_prompt: |
      Você é agente de nurturing B2B. Para cada lead:
      1. Busque contexto via hubspot-mcp (contact + engagements)
      2. Calcule score com regras em nurture-rules.md (NÃO invente pesos)
      3. Consulte estado em nurture.db via sqlite-mcp
      4. Se HANDOFF_SDR: gere brief JSON, NÃO escreva email
      5. Se SEND_*: copy em 3 passos (outline → draft → critique)
      6. Valide guardrails em guardrails.md
      7. NUNCA envie email — submeta para aprovação Slack
      8. Registre tudo em audit_log
    tools:
      - hubspot-mcp
      - sqlite-mcp
      - slack-mcp
    approval_required:
      - send_email
      - update_crm

Arquivo nurture-rules.md (versionado no repo):

# Regras de Scoring — v1.2
- Comportamental 40%, ICP 30%, Recência 30%
- HOT (>=75): HANDOFF_SDR, sem email
- Rate limit: 1 email / 5 dias / lead
- Opt-out: STOP imediato

12 · Avaliação: Golden Set + LLM-as-Judge

Antes de produção, congele um golden set de 10 leads (anonimizados) com decisão esperada:

def run_regression(fixtures: list[dict], pipeline) -> dict:
    results = []
    for f in fixtures:
        out = pipeline(f["contact_id"])
        results.append({
            "fixture": f["name"],
            "expected_action": f["expected_action"],
            "actual_action": out["action"],
            "match": out["action"] == f["expected_action"],
        })
    accuracy = sum(r["match"] for r in results) / len(results)
    return {"accuracy": accuracy, "results": results}

Para qualidade de copy, use LLM-as-judge com rubrica (conecta com métricas de avaliação):

13 · Casos Reais com Números

Transparência: cenários baseados em implementações B2B SaaS típicas. Não são auditoria independente — use como ordem de magnitude.

Caso 1 — Base de 180 MQLs, SaaS martech

Insight: volume de email caiu 60%, resposta subiu 4.5x. Menos email, melhor email.

Caso 2 — Handoff SDR para leads HOT

Antes: leads com visita em /pricing esperavam média 3.2 dias até contato humano (fila do SDR). Depois: webhook + brief automático → SLA 4h → tempo médio 2.8 horas. Conversão SQL desses leads: +22% (amostra pequena, n=34).

Caso 3 — Custo operacional

14 · Métricas e Dashboard

# Métricas semanais — query SQLite
WEEKLY_METRICS = """
SELECT
  COUNT(*) FILTER (WHERE event='sent') as emails_sent,
  COUNT(*) FILTER (WHERE event='reply') as replies,
  COUNT(*) FILTER (WHERE event='handoff') as handoffs,
  COUNT(*) FILTER (WHERE event='skip') as skipped,
  ROUND(100.0 * COUNT(*) FILTER (WHERE event='reply') /
    NULLIF(COUNT(*) FILTER (WHERE event='sent'), 0), 1) as reply_rate_pct
FROM audit_log
WHERE created_at >= date('now', '-7 days');
"""

Skip rate alto é saudável — significa que o agente está filtrando, não spammando.

15 · LGPD e Compliance

Consulte seu DPO antes de automatizar envio — mesmo com aprovação humana, você é responsável pelo conteúdo.

16 · Armadilhas

1. Scoring só com LLM — impossível debugar; use regras primeiro
2. Email para lead HOT — perde timing; handoff imediato
3. Sem rate limit — 3 emails em 10 dias = opt-out
4. Copy em 1 passo — genérico; use 3 passos
5. Ignorar webhooks — agente não aprende com respostas
6. CRM sujo — agente alucina contexto; garbage in, garbage out
7. Aprovação sem SLA — rascunhos expiram, lead esfria
8. Métrica de volume — KPI errado; meça reply rate e MQL→SQL
9. Automação total dia 1 — queime a base; 8 semanas de gate mínimo
10. Sem golden set — regressão silenciosa a cada mudança de prompt

17 · Rollout em 7 Semanas

Conclusão

Agente de nurturing é um sistema de decisão com copy como output — não um gerador de email. A sequência importa: CRM rico → scoring explicável → FSM → copy 3-pass → guardrails → aprovação → envio → webhook → feedback.

O que fazer agora:

1. Clone a estrutura nurture-agent/ e configure HUBSPOT_ACCESS_TOKEN
2. Rode build_lead_context nos 3 perfis da seção 00 — valide se as decisões fazem sentido
3. Implemente scoring sem LLM; só adicione LLM no copy e no classificador de respostas
4. Congele golden set antes de mudar prompts
5. Mantenha approval gate por 8 semanas mínimo

Quando reply rate superar 8% com edit rate abaixo de 10%, você tem um agente de produção — não um experimento.

Publicado em andrecosta.ia.br.