> ## Documentation Index
> Fetch the complete documentation index at: https://docs.supersonic.cv/llms.txt
> Use this file to discover all available pages before exploring further.

# Deal stage notifications

> A Python script that detects deal stage changes and sends Slack notifications. Polls every 5 minutes via cron.

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

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

## The script

```python theme={null}
#!/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

<Steps>
  <Step title="Install dependencies">
    ```bash theme={null}
    pip install httpx
    ```
  </Step>

  <Step title="Set environment variables">
    ```bash theme={null}
    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"
    ```
  </Step>

  <Step title="Run once to initialize state">
    The first run creates the state file. No notifications are sent since there's nothing to compare against.

    ```bash theme={null}
    python deal_notifications.py
    ```

    Check that the state file was created:

    ```bash theme={null}
    cat ~/.supersonic_deal_state.json
    ```
  </Step>

  <Step title="Schedule with cron">
    Run every 5 minutes:

    ```cron theme={null}
    */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
    ```
  </Step>
</Steps>

## Sending email instead of Slack

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

```python theme={null}
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)
```

<Note>
  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.
</Note>

<Tip>
  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:

  ```python theme={null}
  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])
  ```
</Tip>