CRM on autopilot: HubSpot + AI, deals without a sales rep

# HubSpot through 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()

    # Step 1: Claude extracts structured data from the email
    extraction = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Extract the following data from the customer's email as JSON:
            - contact_name: the customer's name
            - company: the company
            - interest: what exactly they're interested in
            - budget_hint: any hint about budget (if there is one)
            - next_step: what needs to happen next
            - urgency: urgency (low/medium/high)

            Email: {email_text}

            Return only JSON, no explanations."""
        }]
    )

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

    # Step 2: Update the contact in HubSpot
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json"
    }

    # Update the contact's properties.
    # HubSpot fills in notes_last_contacted itself (read-only), so
    # we save the email summary as a note (see the practice below), not as a property.
    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
    )

    # Step 3: Create a follow-up task if urgency is high
    if data.get("urgency") == "high":
        task_payload = {
            "properties": {
                "hs_task_subject": f"Urgent follow-up: {data.get('contact_name')}",
                "hs_timestamp": str(int(time.time() * 1000)),  # task due date, a required field
                "hs_task_status": "NOT_STARTED",
                "hs_task_priority": "HIGH",
                "hubspot_owner_id": "owner_id_here"
            }
        }
        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"""Rate this lead's readiness to buy on a scale of 0-100.

            Interaction history:
            {contact_history}

            Return JSON:
            {{
                "score": a number from 0 to 100,
                "reasoning": "a short explanation",
                "recommended_action": "what to do right now",
                "hot_signals": ["list of buying signals"],
                "risk_signals": ["list of warning signs"]
            }}"""
        }]
    )

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

# Example usage
history = """
- April 15: first call, they asked about features
- April 18: they sent a list of requirements
- April 22: they asked for a demo
- April 25: after the demo they asked "what's the fastest implementation timeline?"
- April 28: they wrote "our director wants to meet next week"
"""

result = score_lead(history)
# score: 87
# recommended_action: "Confirm the meeting with the director right away and prepare a proposal"

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"""Write a follow-up email to a customer.

            Context:
            - Name: {contact_name}
            - Last contact: {last_interaction}
            - Days without a reply: {days_since_contact}
            - Deal stage: {deal_stage}

            Requirements:
            - Short (3-4 sentences)
            - Not pushy, not pleading
            - Add value (an article, an insight, a case study)
            - One clear call to action
            - English, a professional but friendly tone"""
        }]
    )

    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']}: stage '{d['stage']}', "
        f"last activity {d['days_stale']} days ago, "
        f"amount {d['amount']}"
        for d in deals
    ])

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Analyze the sales pipeline and flag the problem deals.

            Deals:
            {deals_text}

            For each problem deal (stuck for more than 14 days), give:
            1. Diagnosis: why it's stuck
            2. Way forward: 2-3 concrete steps
            3. Recommendation: keep going or close it as lost

            Format: a list with a heading for each deal."""
        }]
    )

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

# .env
HUBSPOT_ACCESS_TOKEN=<token from 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
):
    """Processes an incoming email and updates the CRM"""

    print(f"Processing email from: {email_from}")

    # 1. Claude analyzes the email
    analysis = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Analyze this incoming customer email.

Subject: {email_subject}
From: {email_from}
Body: {email_body}

Return JSON with no extra text:
{{
    "intent": "inquiry/complaint/purchase_ready/follow_up/other",
    "summary": "a short 1-2 sentence summary",
    "key_points": ["list of important points"],
    "urgency": "low/medium/high",
    "sentiment": "positive/neutral/negative",
    "next_action": "a specific next action",
    "follow_up_days": number of days to wait before writing if there's no reply
}}"""
        }]
    )

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

    # 2. Look up the contact in HubSpot by 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. Update the contact
        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. Add a note
        note_response = requests.post(
            "https://api.hubapi.com/crm/v3/objects/notes",
            headers=headers,
            json={
                "properties": {
                    "hs_note_body": f"Email analysis:\n{data['summary']}\n\nKey points:\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. Create a task if urgency is high
        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"[HOT] Reply to: {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"Created an urgent task for {contact_name}")

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

    else:
        # Contact not found: create a new one
        print(f"New contact: {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}


# Test
if __name__ == "__main__":
    result = process_incoming_email(
        email_from="john@example.com",
        email_subject="We'd like to roll out your system",
        email_body="""
        Hi there,

        We're looking at several options for sales automation.
        Your product caught our interest. We have a team of 15 sales reps,
        the budget is open for discussion, and we'd like to get started this quarter.

        Could we set up a demo next week?

        John Miller, Director of Sales
        """
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

python email_processor.py