Signal Provider → Webhook → Platform Event → Flow → Task/Alert/Field Update

# webhook_receiver.py — Flask app that publishes Salesforce Platform Events
import os
import hmac
import hashlib
from flask import Flask, request, jsonify
from simple_salesforce import Salesforce

app = Flask(__name__)

WEBHOOK_SECRET = os.environ["SIGNAL_WEBHOOK_SECRET"]
SF_USERNAME = os.environ["SF_USERNAME"]
SF_PASSWORD = os.environ["SF_PASSWORD"]
SF_TOKEN = os.environ["SF_SECURITY_TOKEN"]

def verify_signature(payload, signature):
    """Verify webhook signature to prevent spoofing."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhook/signals", methods=["POST"])
def receive_signal():
    # Verify webhook authenticity
    signature = request.headers.get("X-Signature-256", "")
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401

    payload = request.json
    
    # Connect to Salesforce
    sf = Salesforce(
        username=SF_USERNAME,
        password=SF_PASSWORD,
        security_token=SF_TOKEN
    )
    
    # Publish Platform Event
    sf.Signal_Received__e.create({
        "Company_ID__c": payload["company_id"],
        "Signal_Type__c": payload["signal_type"],
        "Signal_Subtype__c": payload["signal_subtype"],
        "Signal_Data__c": json.dumps(payload["data_json"]),
        "Signal_Timestamp__c": payload["timestamp"],
        "Account_Domain__c": payload.get("domain", "")
    })
    
    return jsonify({"status": "accepted"}), 202

if __name__ == "__main__":
    app.run(port=5000)

Signal Provider → Webhook → Custom Code Action → Property Update → Workflow Trigger

// HubSpot Custom Code Action — process incoming signal
const hubspot = require('@hubspot/api-client');

exports.main = async (event, callback) => {
  const hubspotClient = new hubspot.Client({ 
    accessToken: process.env.HUBSPOT_ACCESS_TOKEN 
  });
  
  const signal = event.inputFields;
  
  // Find company by domain
  const searchResponse = await hubspotClient.crm.companies.searchApi.doSearch({
    filterGroups: [{
      filters: [{
        propertyName: 'domain',
        operator: 'EQ',
        value: signal.company_domain
      }]
    }],
    properties: ['domain', 'name', 'hubspot_owner_id']
  });
  
  if (searchResponse.results.length === 0) {
    // No matching company — create one or skip
    callback({ outputFields: { status: 'no_match' } });
    return;
  }
  
  const company = searchResponse.results[0];
  
  // Update company properties
  await hubspotClient.crm.companies.basicApi.update(company.id, {
    properties: {
      'last_signal_type': signal.signal_type,
      'last_signal_date': new Date().toISOString().split('T')[0],
      'signal_score': calculateScore(signal.signal_type, signal.signal_subtype)
    }
  });
  
  // Create engagement (note) for visibility
  await hubspotClient.crm.objects.basicApi.create('notes', {
    properties: {
      hs_note_body: formatSignalNote(signal),
      hs_timestamp: Date.now()
    },
    associations: [{
      to: { id: company.id },
      types: [{ associationCategory: 'HUBSPOT_DEFINED', associationTypeId: 190 }]
    }]
  });
  
  callback({ outputFields: { status: 'processed', company_id: company.id } });
};

function calculateScore(type, subtype) {
  const scores = {
    'funding': 85,
    'job_change': 70,
    'tech_install': 60,
    'intent': 50,
    'hiring': 45
  };
  return scores[type] || 30;
}

function formatSignalNote(signal) {
  return `🔔 Signal Detected: ${signal.signal_type} / ${signal.signal_subtype}\n` +
    `Detected: ${signal.timestamp}\n` +
    `Details: ${JSON.stringify(signal.data_json, null, 2)}`;
}

# hubspot_signal_handler.py
import requests
import os

HUBSPOT_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
BASE_URL = "https://api.hubapi.com"

def process_signal(signal_payload):
    """Route a signal to the matching HubSpot company."""
    
    # Search for company by domain
    search_body = {
        "filterGroups": [{
            "filters": [{
                "propertyName": "domain",
                "operator": "EQ", 
                "value": signal_payload["domain"]
            }]
        }],
        "properties": ["domain", "name", "hubspot_owner_id"]
    }
    
    resp = requests.post(
        f"{BASE_URL}/crm/v3/objects/companies/search",
        json=search_body,
        headers={"Authorization": f"Bearer {HUBSPOT_TOKEN}"}
    )
    
    results = resp.json().get("results", [])
    if not results:
        return {"status": "no_match"}
    
    company_id = results[0]["id"]
    
    # Update signal properties
    requests.patch(
        f"{BASE_URL}/crm/v3/objects/companies/{company_id}",
        json={"properties": {
            "last_signal_type": signal_payload["signal_type"],
            "last_signal_date": signal_payload["timestamp"][:10],
            "signal_score": str(score_signal(signal_payload))
        }},
        headers={"Authorization": f"Bearer {HUBSPOT_TOKEN}"}
    )
    
    return {"status": "processed", "company_id": company_id}

Autobound Signal API → Your Webhook Endpoint → Signal Processor → Custom CRM API

# signal_processor.py — Production webhook receiver with queue
import os
import json
import hmac
import hashlib
import logging
from datetime import datetime, timedelta
from flask import Flask, request, jsonify
from redis import Redis
from rq import Queue

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration
WEBHOOK_SECRET = os.environ["SIGNAL_WEBHOOK_SECRET"]
redis_conn = Redis(host=os.environ.get("REDIS_HOST", "localhost"))
signal_queue = Queue("signals", connection=redis_conn)

# Deduplication cache (prevents processing same signal twice)
DEDUP_TTL = 86400  # 24 hours

def verify_webhook(payload_bytes, signature):
    """HMAC-SHA256 verification of webhook authenticity."""
    computed = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={computed}", signature)

def is_duplicate(signal_id):
    """Check if we've already processed this signal."""
    key = f"signal:seen:{signal_id}"
    if redis_conn.exists(key):
        return True
    redis_conn.setex(key, DEDUP_TTL, "1")
    return False

@app.route("/webhooks/signals", methods=["POST"])
def receive_signal():
    """Webhook endpoint — validates, deduplicates, and enqueues signals."""
    
    # 1. Verify signature
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_webhook(request.data, signature):
        logger.warning("Invalid webhook signature rejected")
        return jsonify({"error": "unauthorized"}), 401
    
    # 2. Parse payload
    payload = request.json
    signal_id = payload.get("signal_id", "")
    
    # 3. Deduplicate
    if is_duplicate(signal_id):
        logger.info(f"Duplicate signal {signal_id} — skipping")
        return jsonify({"status": "duplicate"}), 200
    
    # 4. Enqueue for async processing
    signal_queue.enqueue(
        "workers.process_signal",
        payload,
        job_timeout="5m",
        retry=3
    )
    
    logger.info(f"Signal {signal_id} enqueued: {payload['signal_type']}/{payload['signal_subtype']}")
    return jsonify({"status": "accepted", "signal_id": signal_id}), 202


# --- Worker that processes queued signals ---
# workers.py

import requests
from datetime import datetime

CRM_API_BASE = os.environ["CRM_API_BASE_URL"]
CRM_API_KEY = os.environ["CRM_API_KEY"]

SIGNAL_PRIORITY = {
    "funding": 90,
    "job_change": 80,
    "acquisition": 85,
    "tech_install": 60,
    "hiring_surge": 55,
    "intent_topic": 50,
    "news_mention": 30
}

def process_signal(payload):
    """Process a single signal and route to CRM."""
    
    signal_type = payload["signal_type"]
    company_domain = payload.get("domain", "")
    company_id = payload.get("company_id", "")
    
    # 1. Resolve company in CRM
    crm_account = find_crm_account(company_domain, company_id)
    if not crm_account:
        logger.info(f"No CRM match for {company_domain} — storing for later")
        store_unmatched_signal(payload)
        return
    
    # 2. Calculate priority
    priority = SIGNAL_PRIORITY.get(signal_type, 40)
    
    # 3. Route based on signal type and priority
    if priority >= 80:
        # High priority: create task + alert
        create_crm_task(crm_account, payload, urgent=True)
        send_rep_alert(crm_account["owner_email"], payload)
    elif priority >= 50:
        # Medium priority: create task
        create_crm_task(crm_account, payload, urgent=False)
    else:
        # Low priority: update account intelligence field
        update_account_signals(crm_account, payload)
    
    # 4. Always update the account's signal history
    append_signal_history(crm_account["id"], payload)

def find_crm_account(domain, external_id):
    """Find account in CRM by domain or external ID."""
    headers = {"Authorization": f"Bearer {CRM_API_KEY}"}
    
    # Try external ID first (fastest)
    if external_id:
        resp = requests.get(
            f"{CRM_API_BASE}/accounts",
            params={"external_id": external_id},
            headers=headers
        )
        if resp.status_code == 200 and resp.json().get("data"):
            return resp.json()["data"][0]
    
    # Fall back to domain match
    if domain:
        resp = requests.get(
            f"{CRM_API_BASE}/accounts",
            params={"domain": domain},
            headers=headers
        )
        if resp.status_code == 200 and resp.json().get("data"):
            return resp.json()["data"][0]
    
    return None

def create_crm_task(account, signal, urgent=False):
    """Create a task/activity in the CRM."""
    headers = {"Authorization": f"Bearer {CRM_API_KEY}"}
    
    task_body = {
        "account_id": account["id"],
        "assigned_to": account["owner_id"],
        "subject": format_task_subject(signal),
        "description": format_task_body(signal),
        "priority": "high" if urgent else "normal",
        "due_date": (datetime.now() + timedelta(hours=24 if urgent else 72)).isoformat(),
        "source": "signal_integration",
        "metadata": {
            "signal_type": signal["signal_type"],
            "signal_id": signal["signal_id"]
        }
    }
    
    resp = requests.post(
        f"{CRM_API_BASE}/tasks",
        json=task_body,
        headers=headers
    )
    
    if resp.status_code != 201:
        logger.error(f"Failed to create task: {resp.status_code} {resp.text}")
        raise Exception(f"CRM task creation failed: {resp.status_code}")
    
    return resp.json()

def format_task_subject(signal):
    """Human-readable task subject from signal data."""
    templates = {
        "funding": "🚀 {company} raised funding — time-sensitive outreach",
        "job_change": "👤 Key contact change at {company} — re-engage",
        "tech_install": "💻 {company} installed {detail} — relevant to our solution",
        "acquisition": "🏢 {company} acquired/was acquired — org change opportunity",
        "hiring_surge": "📈 {company} hiring aggressively — growth signal"
    }
    template = templates.get(signal["signal_type"], "🔔 New signal for {company}")
    return template.format(
        company=signal.get("company_name", signal.get("domain", "Unknown")),
        detail=signal.get("data_json", {}).get("detail", "")
    )

# batch_sync.py — Pull signal files from GCS bucket and sync to CRM
from google.cloud import storage
import json
import pyarrow.parquet as pq
from datetime import datetime, timedelta

GCS_BUCKET = "your-signal-delivery-bucket"
PROCESSED_PREFIX = "processed/"

def sync_new_signals():
    """Check for new signal files and process them."""
    client = storage.Client()
    bucket = client.bucket(GCS_BUCKET)
    
    # List files from last 24 hours
    cutoff = datetime.now() - timedelta(hours=24)
    blobs = bucket.list_blobs(prefix="signals/")
    
    new_files = [
        b for b in blobs 
        if b.updated > cutoff 
        and not b.name.startswith(PROCESSED_PREFIX)
    ]
    
    for blob in new_files:
        if blob.name.endswith(".parquet"):
            process_parquet_file(blob)
        elif blob.name.endswith(".jsonl"):
            process_jsonl_file(blob)
        
        # Mark as processed
        bucket.rename_blob(blob, f"{PROCESSED_PREFIX}{blob.name}")

def process_parquet_file(blob):
    """Process a Parquet signal file."""
    local_path = f"/tmp/{blob.name.split('/')[-1]}"
    blob.download_to_filename(local_path)
    
    table = pq.read_table(local_path)
    df = table.to_pandas()
    
    # Process in batches of 100
    for i in range(0, len(df), 100):
        batch = df.iloc[i:i+100]
        signals = batch.to_dict(orient="records")
        bulk_upsert_to_crm(signals)
    
    logger.info(f"Processed {len(df)} signals from {blob.name}")

def bulk_upsert_to_crm(signals):
    """Bulk upsert signals to CRM (batch API call)."""
    headers = {"Authorization": f"Bearer {CRM_API_KEY}"}
    
    # Most CRMs support batch operations
    resp = requests.post(
        f"{CRM_API_BASE}/accounts/batch-update",
        json={"records": [
            {
                "match_field": "domain",
                "match_value": s["domain"],
                "properties": {
                    "last_signal_type": s["signal_type"],
                    "last_signal_date": s["timestamp"],
                    "signal_count": {"increment": 1}
                }
            }
            for s in signals
        ]},
        headers=headers
    )
    
    return resp.json()

# poll_signals.py — Periodic poll for new signals
import requests
import time
from datetime import datetime, timezone

SIGNAL_API_BASE = "https://signals.autobound.ai/api/v1"
SIGNAL_API_KEY = os.environ["SIGNAL_API_KEY"]
POLL_INTERVAL = 300  # 5 minutes

def poll_for_signals(last_timestamp=None):
    """Poll the Signal API for new signals since last check."""
    headers = {"Authorization": f"Bearer {SIGNAL_API_KEY}"}
    params = {"limit": 100, "sort": "timestamp_desc"}
    
    if last_timestamp:
        params["since"] = last_timestamp
    
    resp = requests.get(
        f"{SIGNAL_API_BASE}/signals",
        params=params,
        headers=headers
    )
    
    if resp.status_code != 200:
        logger.error(f"Signal API error: {resp.status_code}")
        return []
    
    return resp.json().get("signals", [])

def run_polling_loop():
    """Main polling loop — runs continuously."""
    last_ts = None
    
    while True:
        signals = poll_for_signals(last_ts)
        
        if signals:
            logger.info(f"Received {len(signals)} new signals")
            for signal in signals:
                process_signal(signal)  # Same function from webhook worker
            last_ts = signals[0]["timestamp"]  # Most recent
        
        time.sleep(POLL_INTERVAL)

def should_create_new_task(account_id, signal_type, crm_client):
    """Check for existing open task before creating a new one."""
    existing = crm_client.search_tasks(
        account_id=account_id,
        signal_type=signal_type,
        status="open",
        created_after=(datetime.now() - timedelta(days=7))
    )
    
    if existing:
        # Append new signal data to existing task
        crm_client.update_task(existing[0]["id"], {
            "description": existing[0]["description"] + f"\n\n--- Updated {datetime.now()} ---\n" + format_signal(signal)
        })
        return False
    return True

def calculate_composite_score(signal, account):
    """Score signal based on type, account fit, and pipeline state."""
    base = SIGNAL_PRIORITY.get(signal["signal_type"], 30)
    
    # ICP multiplier
    if account.get("icp_tier") == "A":
        base *= 1.3
    elif account.get("icp_tier") == "B":
        base *= 1.1
    
    # Open opportunity multiplier
    if account.get("has_open_opportunity"):
        base *= 1.5
    
    # Recency decay — signals older than 48h lose value
    signal_age_hours = (datetime.now() - parse(signal["timestamp"])).total_seconds() / 3600
    if signal_age_hours > 48:
        base *= 0.7
    elif signal_age_hours > 168:  # 1 week
        base *= 0.3
    
    return min(int(base), 100)

def check_rep_fatigue(rep_id, signal_priority):
    """Gate signal delivery based on rep's current load."""
    today_count = get_today_signal_count(rep_id)
    unactioned = get_unactioned_tasks(rep_id)
    
    if unactioned > 20:
        # Rep is overwhelmed — only deliver critical signals
        return signal_priority >= 90
    elif today_count >= 5:
        # Daily cap hit — route to digest
        add_to_daily_digest(rep_id, signal)
        return False
    
    return True

{
  "signal_id": "sig_abc123def456",
  "company_id": "comp_789xyz",
  "company_name": "Acme Corp",
  "domain": "acme.com",
  "signal_type": "funding",
  "signal_subtype": "series_c",
  "timestamp": "2026-05-01T14: 30: 00Z",
  "confidence": 0.95,
  "data_json": {
    "amount": 75000000,
    "currency": "USD",
    "lead_investor": "Sequoia Capital",
    "source_url": "https://techcrunch.com/...",
    "round_type": "Series C"
  },
  "related_contacts": [
    {
      "name": "Jane Smith",
      "title": "CFO",
      "linkedin_url": "https://linkedin.com/in/janesmith"
    }
  ]
}