Prerequisites

Python 3.9+ or Node.js 18+
HubSpot private app token with crm.objects.contacts.read, crm.objects.contacts.write, and crm.objects.companies.read scopes
Slack Bot Token (xoxb-...) with chat:write scope
A server or serverless function to receive webhooks (or a cron scheduler for polling)

Why code?

Code gives you full control over the routing hierarchy, company association lookups, and state normalization — with no per-execution cost and no platform limitations. The webhook handler processes each new contact in real time (sub-second), and the territory config lives in your codebase where it's version-controlled and reviewable.

This is the best approach when you need real-time routing at scale. No polling delays, no task limits, no operation credits. The trade-off is that you host and maintain the service yourself — a server or serverless function to receive HubSpot webhooks, plus monitoring to ensure it stays up.

How it works

HubSpot webhook fires on contact.creation events, hitting your endpoint in real time
Webhook handler fetches the contact's associated company for accurate territory/size data
Routing function applies the three-tier hierarchy: existing account owner, enterprise override, territory match
HubSpot PATCH updates the contact's hubspot_owner_id
Slack API DMs the assigned rep with lead details and a HubSpot link

Step 1: Define the territory config

The territory config is the heart of this recipe. Keep it as a separate data structure so sales ops can update it without touching routing logic.

TERRITORY_CONFIG = {
    # Enterprise override (checked first)
    "enterprise_threshold": 1000,
    "enterprise_rep": {
        "owner_id": "444444", "slack_id": "U04DDDD", "name": "Dave (Enterprise)"
    },
 
    # Territory map: state abbreviation -> rep
    "territories": {
        # Northeast
        "NY": {"owner_id": "111111", "slack_id": "U01AAAA", "name": "Alice"},
        "MA": {"owner_id": "111111", "slack_id": "U01AAAA", "name": "Alice"},
        "CT": {"owner_id": "111111", "slack_id": "U01AAAA", "name": "Alice"},
        "NJ": {"owner_id": "111111", "slack_id": "U01AAAA", "name": "Alice"},
        "PA": {"owner_id": "111111", "slack_id": "U01AAAA", "name": "Alice"},
        # West Coast
        "CA": {"owner_id": "222222", "slack_id": "U02BBBB", "name": "Bob"},
        "WA": {"owner_id": "222222", "slack_id": "U02BBBB", "name": "Bob"},
        "OR": {"owner_id": "222222", "slack_id": "U02BBBB", "name": "Bob"},
        # South
        "FL": {"owner_id": "333333", "slack_id": "U03CCCC", "name": "Carol"},
        "GA": {"owner_id": "333333", "slack_id": "U03CCCC", "name": "Carol"},
        "TX": {"owner_id": "333333", "slack_id": "U03CCCC", "name": "Carol"},
    },
 
    # Default fallback
    "default_rep": {
        "owner_id": "555555", "slack_id": "U05EEEE", "name": "Eve (Catch-all)"
    },
}

Step 2: Build the routing function

import requests
import os
 
HUBSPOT_ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
HEADERS = {"Authorization": f"Bearer {HUBSPOT_ACCESS_TOKEN}", "Content-Type": "application/json"}
 
def get_company_for_contact(contact_id):
    """Fetch the primary associated company for a contact."""
    resp = requests.get(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}/associations/companies",
        headers=HEADERS
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    if not results:
        return None
 
    company_id = results[0]["id"]
    resp = requests.get(
        f"https://api.hubapi.com/crm/v3/objects/companies/{company_id}",
        headers=HEADERS,
        params={"properties": "name,numberofemployees,state,country,hubspot_owner_id"}
    )
    resp.raise_for_status()
    return resp.json()
 
def route_contact(contact, company):
    """Determine the right owner based on territory and size rules."""
    config = TERRITORY_CONFIG
 
    # Priority 1: existing account owner
    if company and company.get("properties", {}).get("hubspot_owner_id"):
        return {
            "owner_id": company["properties"]["hubspot_owner_id"],
            "reason": "Existing account owner",
        }
 
    # Priority 2: enterprise override
    employees = int((company or {}).get("properties", {}).get("numberofemployees") or 0)
    if employees >= config["enterprise_threshold"]:
        rep = config["enterprise_rep"]
        return {**rep, "reason": f"Enterprise ({employees} employees)"}
 
    # Priority 3: territory match
    state = (
        (company or {}).get("properties", {}).get("state")
        or contact.get("properties", {}).get("state")
        or ""
    ).upper()
 
    if state in config["territories"]:
        rep = config["territories"][state]
        return {**rep, "reason": f"Territory: {state}"}
 
    # Fallback
    rep = config["default_rep"]
    return {**rep, "reason": f"No match (state: {state or 'unknown'})"}

Step 3: Wire it together with a webhook handler

from flask import Flask, request, jsonify
from slack_sdk import WebClient
 
app = Flask(__name__)
slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
PORTAL_ID = os.environ.get("HUBSPOT_PORTAL_ID", "YOUR_PORTAL_ID")
 
@app.route("/webhook", methods=["POST"])
def handle_webhook():
    for event in request.json:
        if event.get("subscriptionType") != "contact.creation":
            continue
 
        contact_id = event["objectId"]
 
        # Fetch contact
        resp = requests.get(
            f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
            headers=HEADERS,
            params={"properties": "firstname,lastname,email,company,jobtitle,state,country,hubspot_owner_id"}
        )
        resp.raise_for_status()
        contact = resp.json()
 
        # Skip already-assigned
        if contact["properties"].get("hubspot_owner_id"):
            continue
 
        # Get company
        company = get_company_for_contact(contact_id)
 
        # Route
        result = route_contact(contact, company)
 
        # Assign owner
        requests.patch(
            f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
            headers=HEADERS,
            json={"properties": {"hubspot_owner_id": result["owner_id"]}}
        ).raise_for_status()
 
        # Notify in Slack
        name = f"{contact['properties'].get('firstname', '')} {contact['properties'].get('lastname', '')}".strip()
        company_name = (company or {}).get("properties", {}).get("name", contact["properties"].get("company", "Unknown"))
 
        if result.get("slack_id"):
            slack.chat_postMessage(
                channel=result["slack_id"],
                text=f"New lead routed: {name}",
                blocks=[
                    {"type": "section", "text": {"type": "mrkdwn",
                        "text": f"🆕 *New Lead Routed to You*\n*{name}* at {company_name}\n📍 {result['reason']}"}},
                    {"type": "actions", "elements": [
                        {"type": "button", "text": {"type": "plain_text", "text": "View in HubSpot"},
                         "url": f"https://app.hubspot.com/contacts/{PORTAL_ID}/contact/{contact_id}"}
                    ]}
                ]
            )
 
    return jsonify({"status": "ok"}), 200

Step 4: Register the webhook

curl -X POST "https://api.hubapi.com/webhooks/v3/YOUR_APP_ID/subscriptions" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"eventType": "contact.creation", "active": true}'

Existing account lookup adds latency

Each routed contact requires 2-3 extra API calls (association lookup + company fetch). For high-volume scenarios, cache company data locally or use HubSpot's batch endpoints.

Troubleshooting

Common questions

How do I handle HubSpot webhook retries without double-assigning?

HubSpot retries webhooks if it doesn't receive a 200 response within 5 seconds. If your handler takes longer (multiple API calls per contact), return 200 immediately and process asynchronously using a queue (Bull, SQS, or a simple in-memory set). Add idempotency by checking if the contact already has an owner before assigning.

What's the HubSpot API rate limit for private apps?

100 requests per 10 seconds. Each routed lead requires 3-4 API calls (contact fetch, association lookup, company fetch, owner update). At sustained volume, that's ~25-33 leads per 10-second window. For higher throughput, use HubSpot's batch endpoints to update multiple contacts in a single request.

How do I update territory mappings without redeploying?

Store the territory config in a database, Google Sheet, or environment variable (JSON string) instead of hardcoding it. Fetch at startup or on each request. For a serverless function, environment variable changes take effect on the next cold start — or use a cache with a short TTL.

Can I test this locally before deploying?

Yes. Use ngrok to expose your local server to the internet, then register the ngrok URL as your HubSpot webhook endpoint. Create a test contact in HubSpot and verify the webhook fires, the routing logic runs, and the owner is set correctly. Remove the ngrok subscription before going to production.

Cost

Hosting: Free on Vercel (serverless), ~$5/mo on Railway
No per-execution cost beyond hosting

Route HubSpot leads by territory and company size using code