El CRM en piloto automático: HubSpot + IA, negocios que avanzan sin un vendedor encima

# HubSpot a través de Claude + API
import os
import time
import anthropic
import requests

HUBSPOT_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]

def update_crm_from_email(email_text: str, contact_id: str):
    client = anthropic.Anthropic()

    # Paso 1: Claude extrae datos estructurados del correo
    extraction = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Extrae del correo del cliente los siguientes datos en JSON:
            - contact_name: nombre del cliente
            - company: empresa
            - interest: qué le interesa concretamente
            - budget_hint: pista sobre el presupuesto (si la hay)
            - next_step: qué hay que hacer después
            - urgency: urgencia (low/medium/high)

            Correo: {email_text}

            Devuelve solo el JSON, sin explicaciones."""
        }]
    )

    import json
    data = json.loads("".join(b.text for b in extraction.content if b.type == "text"))

    # Paso 2: actualizamos el contacto en HubSpot
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json"
    }

    # Actualizamos las propiedades del contacto.
    # HubSpot llena por su cuenta el campo notes_last_contacted (solo lectura), así que
    # el resumen del correo lo guardamos como nota (ver la práctica abajo), no como propiedad.
    update_payload = {
        "properties": {
            "hs_lead_status": "IN_PROGRESS"
        }
    }

    response = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
        headers=headers,
        json=update_payload
    )

    # Paso 3: creamos una tarea de seguimiento si la urgencia es alta
    if data.get("urgency") == "high":
        task_payload = {
            "properties": {
                "hs_task_subject": f"Seguimiento urgente: {data.get('contact_name')}",
                "hs_timestamp": str(int(time.time() * 1000)),  # fecha límite de la tarea, campo obligatorio
                "hs_task_status": "NOT_STARTED",
                "hs_task_priority": "HIGH",
                "hubspot_owner_id": "owner_id_aqui"
            }
        }
        requests.post(
            "https://api.hubapi.com/crm/v3/objects/tasks",
            headers=headers,
            json=task_payload
        )

    return data

def score_lead(contact_history: str) -> dict:
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"""Califica la disposición de compra del prospecto en una escala de 0 a 100.

            Historial de interacciones:
            {contact_history}

            Devuelve JSON:
            {{
                "score": número de 0 a 100,
                "reasoning": "explicación breve",
                "recommended_action": "qué hacer ahora mismo",
                "hot_signals": ["lista de señales de disposición"],
                "risk_signals": ["lista de señales de alerta"]
            }}"""
        }]
    )

    import json
    return json.loads("".join(b.text for b in response.content if b.type == "text"))

# Ejemplo de uso
history = """
- 15 de abril: primera llamada, preguntaron por las funciones
- 18 de abril: mandaron su lista de requisitos
- 22 de abril: pidieron una demo
- 25 de abril: después de la demo preguntaron "¿cuál es el plazo mínimo de implementación?"
- 28 de abril: escribieron "nuestro director quiere reunirse la próxima semana"
"""

result = score_lead(history)
# score: 87
# recommended_action: "Confirmar de inmediato la reunión con el director, preparar la propuesta comercial"

def generate_followup(
    contact_name: str,
    last_interaction: str,
    days_since_contact: int,
    deal_stage: str
) -> str:
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": f"""Escribe un correo de seguimiento para el cliente.

            Contexto:
            - Nombre: {contact_name}
            - Último contacto: {last_interaction}
            - Días sin respuesta: {days_since_contact}
            - Etapa del negocio: {deal_stage}

            Requisitos:
            - Corto (3-4 oraciones)
            - Sin presionar y sin suplicar
            - Aporta valor (un artículo, una idea, un caso)
            - Un solo llamado a la acción claro
            - En español, tono profesional pero cercano"""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")

def audit_pipeline(deals: list) -> list:
    client = anthropic.Anthropic()

    deals_text = "\n".join([
        f"- {d['name']}: etapa '{d['stage']}', "
        f"última acción hace {d['days_stale']} días, "
        f"monto {d['amount']}"
        for d in deals
    ])

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Analiza el pipeline de ventas y señala los negocios con problemas.

            Negocios:
            {deals_text}

            Para cada negocio con problemas (atorado más de 14 días) da:
            1. Diagnóstico: por qué se atoró
            2. Plan para destrabarlo: 2-3 pasos concretos
            3. Recomendación: seguir o cerrarlo como perdido

            Formato: una lista con un encabezado por cada negocio."""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")

# .env
HUBSPOT_ACCESS_TOKEN=<token de HubSpot>
ANTHROPIC_API_KEY=sk-ant-...

# email_processor.py
import os
import json
import time
import anthropic
import requests
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
HUBSPOT_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]

def process_incoming_email(
    email_from: str,
    email_subject: str,
    email_body: str
):
    """Procesa un correo entrante y actualiza el CRM"""

    print(f"Procesando correo de: {email_from}")

    # 1. Claude analiza el correo
    analysis = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Analiza el correo entrante del cliente.

Asunto: {email_subject}
De: {email_from}
Texto: {email_body}

Devuelve JSON sin texto adicional:
{{
    "intent": "inquiry/complaint/purchase_ready/follow_up/other",
    "summary": "resumen breve en 1-2 oraciones",
    "key_points": ["lista de puntos importantes"],
    "urgency": "low/medium/high",
    "sentiment": "positive/neutral/negative",
    "next_action": "siguiente acción concreta",
    "follow_up_days": número de días tras los cuales escribir si no hay respuesta
}}"""
        }]
    )

    data = json.loads("".join(b.text for b in analysis.content if b.type == "text"))
    print(f"Análisis: {data['intent']}, urgencia: {data['urgency']}")

    # 2. Buscamos el contacto en HubSpot por email
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json"
    }

    search_response = requests.post(
        "https://api.hubapi.com/crm/v3/objects/contacts/search",
        headers=headers,
        json={
            "filterGroups": [{
                "filters": [{
                    "propertyName": "email",
                    "operator": "EQ",
                    "value": email_from
                }]
            }],
            "properties": ["email", "firstname", "lastname", "hs_lead_status"]
        }
    )

    contacts = search_response.json().get("results", [])

    if contacts:
        contact_id = contacts[0]["id"]
        contact_name = (
            contacts[0]["properties"].get("firstname", "") + " " +
            contacts[0]["properties"].get("lastname", "")
        ).strip() or email_from

        # 3. Actualizamos el contacto
        update_response = requests.patch(
            f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
            headers=headers,
            json={
                "properties": {
                    "hs_lead_status": "IN_PROGRESS" if data["intent"] == "purchase_ready" else "OPEN"
                }
            }
        )

        # 4. Agregamos una nota
        note_response = requests.post(
            "https://api.hubapi.com/crm/v3/objects/notes",
            headers=headers,
            json={
                "properties": {
                    "hs_note_body": f"Análisis del correo:\n{data['summary']}\n\nPuntos clave:\n" +
                                   "\n".join(f"• {p}" for p in data["key_points"]),
                    "hs_timestamp": str(int(time.time() * 1000))
                },
                "associations": [{
                    "to": {"id": contact_id},
                    "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": 202}]
                }]
            }
        )

        # 5. Creamos una tarea si la urgencia es alta
        if data["urgency"] == "high" or data["intent"] == "purchase_ready":
            task_response = requests.post(
                "https://api.hubapi.com/crm/v3/objects/tasks",
                headers=headers,
                json={
                    "properties": {
                        "hs_task_subject": f"[CALIENTE] Responder: {contact_name} — {email_subject}",
                        "hs_timestamp": str(int(time.time() * 1000)),
                        "hs_task_status": "NOT_STARTED",
                        "hs_task_priority": "HIGH",
                        "hs_task_body": data["next_action"]
                    }
                }
            )
            print(f"Tarea urgente creada para {contact_name}")

        print(f"CRM actualizado para {contact_name}")
        return {"status": "updated", "contact": contact_name, "analysis": data}

    else:
        # No se encontró el contacto: creamos uno nuevo
        print(f"Contacto nuevo: {email_from}")
        create_response = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts",
            headers=headers,
            json={
                "properties": {
                    "email": email_from,
                    "hs_lead_status": "NEW"
                }
            }
        )
        return {"status": "created", "contact": email_from, "analysis": data}


# Prueba
if __name__ == "__main__":
    result = process_incoming_email(
        email_from="[email protected]",
        email_subject="Queremos implementar su sistema",
        email_body="""
        ¡Buenas tardes!

        Estamos evaluando varias soluciones para automatizar ventas.
        Su producto nos interesó. Tenemos un equipo de 15 vendedores,
        el presupuesto está por definirse y nos gustaría empezar este mismo trimestre.

        ¿Podríamos tener una demo la próxima semana?

        Iván Pérez, Director de Ventas
        """
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

python email_processor.py