How to Integrate Replit with HubSpot Marketing Hub

1import os
2import requests
3from typing import Optional
4
5ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
6PORTAL_ID = os.environ["HUBSPOT_PORTAL_ID"]
7BASE_URL = "https://api.hubapi.com"
8
9HEADERS = {
10    "Authorization": f"Bearer {ACCESS_TOKEN}",
11    "Content-Type": "application/json"
12}
13
14def get_contact_by_email(email: str) -> Optional[dict]:
15    """Look up a contact by email address."""
16    url = f"{BASE_URL}/crm/v3/objects/contacts/{email}"
17    params = {
18        "idProperty": "email",
19        "properties": "firstname,lastname,email,company,hs_lead_status,lifecyclestage"
20    }
21    response = requests.get(url, headers=HEADERS, params=params)
22    if response.status_code == 404:
23        return None
24    response.raise_for_status()
25    return response.json()
26
27def create_or_update_contact(
28    email: str,
29    first_name: str = "",
30    last_name: str = "",
31    company: str = "",
32    extra_properties: dict = None
33) -> dict:
34    """
35    Upsert a contact in HubSpot.
36    If the email exists, updates properties. If not, creates a new contact.
37    """
38    properties = {
39        "email": email,
40        "firstname": first_name,
41        "lastname": last_name,
42        "company": company,
43    }
44    if extra_properties:
45        properties.update(extra_properties)
46
47    # Try to create first
48    response = requests.post(
49        f"{BASE_URL}/crm/v3/objects/contacts",
50        json={"properties": properties},
51        headers=HEADERS
52    )
53
54    # If contact exists (409 Conflict), update instead
55    if response.status_code == 409:
56        existing = get_contact_by_email(email)
57        if existing:
58            contact_id = existing['id']
59            update_response = requests.patch(
60                f"{BASE_URL}/crm/v3/objects/contacts/{contact_id}",
61                json={"properties": properties},
62                headers=HEADERS
63            )
64            update_response.raise_for_status()
65            return update_response.json()
66    
67    response.raise_for_status()
68    return response.json()
69
70def add_contact_to_list(contact_id: str, list_id: str) -> bool:
71    """Add a contact to a HubSpot static list."""
72    url = f"{BASE_URL}/contacts/v1/lists/{list_id}/add"
73    payload = {"vids": [int(contact_id)]}
74    response = requests.post(url, json=payload, headers=HEADERS)
75    response.raise_for_status()
76    return True
77
78def enroll_in_workflow(contact_email: str, workflow_id: str) -> bool:
79    """Enroll a contact in a HubSpot workflow by email."""
80    url = f"{BASE_URL}/automation/v4/enrollment/automations/{workflow_id}/enrollments/contacts"
81    payload = {"email": contact_email}
82    response = requests.post(url, json=payload, headers=HEADERS)
83    response.raise_for_status()
84    return True
85
86def get_marketing_emails(limit: int = 20) -> list:
87    """Fetch marketing email campaigns and their stats."""
88    url = f"{BASE_URL}/marketing/v3/emails"
89    params = {"limit": limit, "orderBy": "-stats.sent"}
90    response = requests.get(url, headers=HEADERS, params=params)
91    response.raise_for_status()
92    return response.json().get('results', [])
93
94# Example usage
95if __name__ == "__main__":
96    # Create a test contact
97    contact = create_or_update_contact(
98        email="test@example.com",
99        first_name="Jane",
100        last_name="Doe",
101        company="Acme Corp",
102        extra_properties={"lifecyclestage": "lead"}
103    )
104    print(f"Contact: {contact['id']} — {contact['properties'].get('email')}")
105
106    # Fetch email campaigns
107    emails = get_marketing_emails(5)
108    print(f"\nMarketing emails: {[e.get('name', 'Unnamed') for e in emails]}")

1const express = require('express');
2const hubspot = require('@hubspot/api-client');
3
4const app = express();
5app.use(express.json());
6
7const ACCESS_TOKEN = process.env.HUBSPOT_ACCESS_TOKEN;
8const WELCOME_WORKFLOW_ID = process.env.HUBSPOT_ONBOARDING_WORKFLOW_ID || '';
9
10// Initialize HubSpot client
11const hubspotClient = new hubspot.Client({ accessToken: ACCESS_TOKEN });
12
13// Create or update a contact (upsert by email)
14app.post('/leads', async (req, res) => {
15  const { email, firstName, lastName, company, phone, source } = req.body;
16  if (!email) return res.status(400).json({ error: 'email is required' });
17
18  try {
19    const properties = {
20      email,
21      firstname: firstName || '',
22      lastname: lastName || '',
23      company: company || '',
24      phone: phone || '',
25      lead_source: source || 'web-form',
26      lifecyclestage: 'lead'
27    };
28
29    let contactId;
30    try {
31      // Try to create
32      const createResponse = await hubspotClient.crm.contacts.basicApi.create({
33        properties
34      });
35      contactId = createResponse.id;
36    } catch (createErr) {
37      if (createErr.code === 409) {
38        // Contact exists — update instead
39        const existing = await hubspotClient.crm.contacts.basicApi.getById(
40          email, ['id'], undefined, undefined, false, 'email'
41        );
42        await hubspotClient.crm.contacts.basicApi.update(existing.id, { properties });
43        contactId = existing.id;
44      } else {
45        throw createErr;
46      }
47    }
48
49    // Optionally enroll in onboarding workflow
50    if (WELCOME_WORKFLOW_ID && contactId) {
51      try {
52        const axios = require('axios');
53        await axios.post(
54          `https://api.hubapi.com/automation/v4/enrollment/automations/${WELCOME_WORKFLOW_ID}/enrollments/contacts`,
55          { email },
56          { headers: { Authorization: `Bearer ${ACCESS_TOKEN}`, 'Content-Type': 'application/json' } }
57        );
58      } catch (workflowErr) {
59        console.warn('Workflow enrollment failed:', workflowErr.message);
60        // Non-critical — contact was still created
61      }
62    }
63
64    res.status(201).json({ success: true, contactId });
65  } catch (err) {
66    console.error('HubSpot error:', err.message);
67    res.status(500).json({ error: err.message });
68  }
69});
70
71// Get recent contacts with lifecycle stage
72app.get('/contacts', async (req, res) => {
73  try {
74    const response = await hubspotClient.crm.contacts.basicApi.getPage(
75      parseInt(req.query.limit) || 20,
76      undefined,
77      ['email', 'firstname', 'lastname', 'lifecyclestage', 'createdate']
78    );
79    res.json(response.results);
80  } catch (err) {
81    res.status(500).json({ error: err.message });
82  }
83});
84
85// Get marketing email list
86app.get('/email-campaigns', async (req, res) => {
87  try {
88    const axios = require('axios');
89    const response = await axios.get(
90      'https://api.hubapi.com/marketing/v3/emails',
91      {
92        headers: { Authorization: `Bearer ${ACCESS_TOKEN}` },
93        params: { limit: 20 }
94      }
95    );
96    res.json(response.data.results || []);
97  } catch (err) {
98    res.status(500).json({ error: err.message });
99  }
100});
101
102app.listen(3000, '0.0.0.0', () => {
103  console.log('HubSpot Marketing Hub integration server running on port 3000');
104});

1from flask import Flask, request, jsonify
2import os
3import hmac
4import hashlib
5import time
6
7app = Flask(__name__)
8
9ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
10CLIENT_SECRET = os.environ.get("HUBSPOT_CLIENT_SECRET", "")
11
12def verify_hubspot_signature(request_body: bytes, timestamp: str, signature: str) -> bool:
13    """
14    Verify HubSpot webhook signature v3.
15    Validates that the request came from HubSpot.
16    """
17    if not CLIENT_SECRET:
18        return True  # Skip verification in development
19    
20    # Check timestamp is within 5 minutes to prevent replay attacks
21    if abs(time.time() - int(timestamp) / 1000) > 300:
22        return False
23    
24    # Build the string to sign: method + URI + body + timestamp
25    # Note: HubSpot uses the full request URI for v3 signatures
26    expected = hmac.new(
27        CLIENT_SECRET.encode(),
28        request_body + timestamp.encode(),
29        hashlib.sha256
30    ).hexdigest()
31    return hmac.compare_digest(expected, signature)
32
33@app.route('/hubspot/webhook', methods=['POST'])
34def hubspot_webhook():
35    # Verify signature
36    signature = request.headers.get('X-HubSpot-Signature-v3', '')
37    timestamp = request.headers.get('X-HubSpot-Request-Timestamp', '0')
38    
39    if CLIENT_SECRET and not verify_hubspot_signature(request.data, timestamp, signature):
40        return jsonify({'error': 'Invalid signature'}), 401
41    
42    events = request.json
43    if not events:
44        return jsonify({'error': 'No events'}), 400
45    
46    for event in events:
47        event_type = event.get('subscriptionType')
48        object_id = event.get('objectId')
49        portal_id = event.get('portalId')
50        
51        print(f"HubSpot event: {event_type} for object {object_id} in portal {portal_id}")
52        
53        if event_type == 'contact.creation':
54            # New contact created — trigger welcome workflow, send to CRM, etc.
55            print(f"New contact: {object_id}")
56        elif event_type == 'contact.propertyChange':
57            property_name = event.get('propertyName')
58            new_value = event.get('propertyValue')
59            print(f"Contact {object_id} property '{property_name}' changed to '{new_value}'")
60    
61    return jsonify({'received': True}), 200
62
63if __name__ == '__main__':
64    app.run(host='0.0.0.0', port=3000)