Deal stage notifications

This script polls your deals pipeline for recently updated entries, compares them against a local state file to detect stage changes, and sends a Slack notification when a deal moves. Run it every 5 minutes with cron.

How it works

Fetch all pipeline entries from Supersonic.
Load the previous state from a local JSON file.
Compare: find entries where the stage changed.
Post a Slack message for each change.
Save the current state for the next run.

The script

#!/usr/bin/env python3
"""
Deal stage change notifier.
Polls pipeline entries, detects stage changes, sends Slack notifications.

Env vars: SUPERSONIC_API_KEY, SLACK_WEBHOOK_URL, PIPELINE_LIST_ID
State file: ~/.supersonic_deal_state.json
"""

import json
import os
from pathlib import Path
import httpx

API_URL = "https://mcp.supersonic.cv/api/developers/mcp/call/"
API_KEY = os.environ["SUPERSONIC_API_KEY"]
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
PIPELINE_LIST_ID = os.environ["PIPELINE_LIST_ID"]
STATE_FILE = Path.home() / ".supersonic_deal_state.json"


def api_call(tool: str, params: dict) -> dict:
    resp = httpx.post(
        API_URL,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={"tool": tool, "params": params},
        timeout=15.0,
    )
    resp.raise_for_status()
    return resp.json()


def get_pipeline_entries() -> list[dict]:
    data = api_call("lists.entries", {"list_id": PIPELINE_LIST_ID})
    return data.get("entries", [])


def load_state() -> dict:
    """Load previous entry states. Returns {entry_id: stage}."""
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}


def save_state(state: dict):
    STATE_FILE.write_text(json.dumps(state, indent=2))


def notify_slack(deal_name: str, old_stage: str, new_stage: str):
    message = f"*Deal moved*: {deal_name}\n{old_stage} -> {new_stage}"
    httpx.post(
        SLACK_WEBHOOK_URL,
        json={"text": message},
        timeout=10.0,
    )


def main():
    entries = get_pipeline_entries()
    old_state = load_state()
    new_state = {}
    changes = []

    for entry in entries:
        entry_id = entry["id"]
        stage = entry.get("data", {}).get("Stage", "Unknown")
        record_data = entry.get("record", {}).get("data", {})
        deal_name = record_data.get("Deal Name") or record_data.get("Name") or entry_id

        new_state[entry_id] = stage

        if entry_id in old_state and old_state[entry_id] != stage:
            changes.append({
                "deal_name": deal_name,
                "old_stage": old_state[entry_id],
                "new_stage": stage,
            })

    # Notify for each change
    for change in changes:
        notify_slack(change["deal_name"], change["old_stage"], change["new_stage"])
        print(f"{change['deal_name']}: {change['old_stage']} -> {change['new_stage']}")

    save_state(new_state)

    if not changes:
        print("No stage changes detected.")
    else:
        print(f"{len(changes)} stage change(s) notified.")


if __name__ == "__main__":
    main()

Setup

Install dependencies

pip install httpx

Set environment variables

export SUPERSONIC_API_KEY="supersonic_live_YOUR_KEY"
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/T00000/B00000/XXXX"
export PIPELINE_LIST_ID="your-pipeline-list-id"

Run once to initialize state

The first run creates the state file. No notifications are sent since there’s nothing to compare against.

python deal_notifications.py

Check that the state file was created:

cat ~/.supersonic_deal_state.json

Schedule with cron

Run every 5 minutes:

*/5 * * * * SUPERSONIC_API_KEY=supersonic_live_YOUR_KEY SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... PIPELINE_LIST_ID=your-id /usr/bin/python3 /path/to/deal_notifications.py

Sending email instead of Slack

Replace the notify_slack function with an email sender. Here’s one using smtplib:

import smtplib
from email.mime.text import MIMEText

def notify_email(deal_name: str, old_stage: str, new_stage: str):
    msg = MIMEText(f"{deal_name} moved from {old_stage} to {new_stage}")
    msg["Subject"] = f"Deal update: {deal_name}"
    msg["From"] = os.environ["SMTP_FROM"]
    msg["To"] = os.environ["NOTIFY_EMAIL"]

    with smtplib.SMTP(os.environ["SMTP_HOST"], int(os.environ.get("SMTP_PORT", 587))) as server:
        server.starttls()
        server.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
        server.send_message(msg)

The state file stores one snapshot of all entry stages. If the script misses a run (e.g., the machine was off), it will detect changes on the next run. It won’t detect intermediate stage changes that happened between runs.

For new deals entering the pipeline, check for entry IDs that exist in the current state but not in the old state. Add this after the comparison loop:

new_entries = set(new_state.keys()) - set(old_state.keys())
for entry_id in new_entries:
    entry = next(e for e in entries if e["id"] == entry_id)
    record_data = entry.get("record", {}).get("data", {})
    deal_name = record_data.get("Deal Name") or record_data.get("Name")
    notify_slack(f"New deal: {deal_name}", "---", new_state[entry_id])

Getting Started

Connect

Examples

Agent Workflows

Sources

Reference

Deal stage notifications

How it works

The script

Setup

Sending email instead of Slack

​How it works

​The script

​Setup

​Sending email instead of Slack

How it works

The script

Setup

Sending email instead of Slack