Automate job change outreach with signal agents

"""Find new job starters with a signal agent and contact each one through a dynamic campaign."""
import json
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")


class FuseError(Exception):
    def __init__(self, status, code, message):
        super().__init__(f"{status} {code}: {message}")
        self.code = code


def call(method, path, **kwargs):
    """Send one request, wait out 429s, and raise FuseError on any other failure."""
    while True:
        response = requests.request(method, BASE + path, auth=AUTH, timeout=60, **kwargs)
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 5)))
            continue
        if response.status_code >= 400:
            try:
                error = response.json().get("error", {})
            except ValueError:
                error = {}
            raise FuseError(response.status_code, error.get("code"), error.get("message"))
        return response.json() if response.content else None


def wait_until(read, done, delay=10, max_delay=60):
    """Poll read() until done(result) is true, waiting before each poll and backing off."""
    while True:
        time.sleep(delay)
        result = read()
        if done(result):
            return result
        delay = min(delay * 1.5, max_delay)


def find_agent(agent_id):
    """GET /agents is the only read that returns an agent's results listId."""
    page = 1
    while True:
        body = call("GET", "/agents", params={"limit": 100, "pageNum": page})
        for agent in body["agents"]:
            if agent["id"] == agent_id:
                return agent
        if page >= body["pagination"]["totalPages"]:
            raise SystemExit(f"Agent {agent_id} not found")
        page += 1


# 1. Resolve the exact values the agent matches on
industry = call("GET", "/agents/autocomplete",
                params={"field": "industry", "q": "financial services"})["suggestions"][0]
titles = call("GET", "/agents/autocomplete",
              params={"field": "title", "q": "chief financial officer"})["suggestions"][:3]
print("Industry:", industry, "| Titles:", titles)

# 2. Save the agent as a draft (nothing is billed yet)
agent_id = call("POST", "/agents/person-starting-new-job", json={
    "listName": "New finance leaders - UK fintech",
    "industries": [industry],
    "authorTitle": titles,
    "leadCompanyHeadcount": ["51-200", "201-500"],
    "leadCompanyCountries": ["United Kingdom"],
    "enrichEmail": True,
    "maxCreditSpend": 5000,
    "expirationDate": None,
})["agent"]["id"]

# 3. Review the stored config, then start the agent
print(json.dumps(call("GET", f"/agents/{agent_id}")["agent"]["config"], indent=2))
if input("Start the agent? It bills 100 credits per person it delivers. [y/N] ").strip().lower() != "y":
    raise SystemExit("The draft is saved and nothing was billed.")
call("POST", f"/agents/{agent_id}/activate")

# 4-5. Wait for the first people to land in the agent's results list
agent = wait_until(
    lambda: find_agent(agent_id),
    lambda current: current["contactsFound"] > 0 or current["status"] in ("failed", "paused", "expired"),
    delay=300,
    max_delay=1800,
)
if not agent["contactsFound"]:
    raise SystemExit(f"The agent stopped with status {agent['status']} before delivering anyone.")
list_id = agent["listId"]
print(f"{agent['contactsFound']} new starters delivered into list {list_id}")

# 6. Point a dynamic AI campaign at the results list
campaign_body = {
    "campaignName": "New finance leaders - welcome",
    "campaignType": "ai-generated",
    "listId": list_id,
    "isDynamic": True,
    "channels": ["email", "email"],
    "ctaLink": "https://example.com/demo",
    "userPrompt": "Congratulate them on the new role. Offer a short call on how new finance leaders "
                  "speed up month-end close in their first 90 days. Warm, brief, no hard sell.",
    "excludeContacted": "last_6_months",
}
for attempt in range(5):
    try:
        campaign_id = call("POST", "/campaigns", json=campaign_body)["campaignId"]
        break
    except FuseError as error:
        if error.code != "LIST_INCOMPATIBLE_WITH_CHANNELS" or attempt == 4:
            raise
        time.sleep(90)

campaign = wait_until(
    lambda: call("GET", f"/campaigns/{campaign_id}")["campaign"],
    lambda current: current["initializationStatus"] in ("completed", "failed"),
)
if campaign["initializationStatus"] == "failed":
    raise SystemExit(campaign["initializationError"])

first_step = call("GET", f"/campaigns/{campaign_id}/steps")["steps"][0]
print(call("POST", f"/campaigns/{campaign_id}/steps/{first_step['id']}/generate")["preview"]["body"])

# 7. Approving sends real email, so ask first
if input("Approve the campaign? [y/N] ").strip().lower() == "y":
    call("POST", f"/campaigns/{campaign_id}/approve")
    print("Live. New starters will be enrolled automatically about every 30 minutes.")