How to Integrate Replit with Autopilot

1import os
2import requests
3
4API_KEY = os.environ["AUTOPILOT_API_KEY"]
5BASE_URL = "https://api2.autopilothq.com/v1"
6
7HEADERS = {
8    "autopilotapikey": API_KEY,
9    "Content-Type": "application/json"
10}
11
12def add_or_update_contact(email: str, first_name: str = "", last_name: str = "",
13                          custom_fields: dict = None) -> dict:
14    """Add a new contact or update an existing one by email."""
15    contact_data = {
16        "contact": {
17            "Email": email,
18            "FirstName": first_name,
19            "LastName": last_name
20        }
21    }
22    # Add custom fields with correct type prefix
23    if custom_fields:
24        for key, value in custom_fields.items():
25            if isinstance(value, bool):
26                contact_data["contact"][f"boolean--{key}"] = value
27            elif isinstance(value, int):
28                contact_data["contact"][f"integer--{key}"] = value
29            elif isinstance(value, float):
30                contact_data["contact"][f"float--{key}"] = value
31            else:
32                contact_data["contact"][f"string--{key}"] = str(value)
33
34    response = requests.post(
35        f"{BASE_URL}/contact",
36        json=contact_data,
37        headers=HEADERS
38    )
39    response.raise_for_status()
40    return response.json()
41
42def get_contact(email: str) -> dict:
43    """Retrieve a contact record by email address."""
44    import urllib.parse
45    encoded_email = urllib.parse.quote(email)
46    response = requests.get(
47        f"{BASE_URL}/contact/{encoded_email}",
48        headers=HEADERS
49    )
50    if response.status_code == 404:
51        return None
52    response.raise_for_status()
53    return response.json()
54
55def enroll_in_journey(journey_id: str, contact_email: str) -> dict:
56    """Trigger a contact's enrollment into a specific journey."""
57    data = {"contact": {"Email": contact_email}}
58    response = requests.post(
59        f"{BASE_URL}/journey/{journey_id}/contact",
60        json=data,
61        headers=HEADERS
62    )
63    response.raise_for_status()
64    return response.json()
65
66def add_to_list(list_id: str, contact_email: str) -> dict:
67    """Add a contact to a smart segment/list."""
68    data = {"contact": {"Email": contact_email}}
69    response = requests.post(
70        f"{BASE_URL}/list/{list_id}/subscriber",
71        json=data,
72        headers=HEADERS
73    )
74    response.raise_for_status()
75    return response.json()
76
77# Example usage
78if __name__ == "__main__":
79    result = add_or_update_contact(
80        email="user@example.com",
81        first_name="Jane",
82        last_name="Doe",
83        custom_fields={
84            "plan_type": "pro",
85            "signup_source": "organic",
86            "trial_days_remaining": 14
87        }
88    )
89    print(f"Contact created/updated: {result}")
90
91    journey_id = os.environ.get("AUTOPILOT_ONBOARDING_JOURNEY_ID", "")
92    if journey_id:
93        enroll_in_journey(journey_id, "user@example.com")
94        print("Enrolled in onboarding journey")

1const express = require('express');
2const axios = require('axios');
3
4const app = express();
5app.use(express.json());
6
7const API_KEY = process.env.AUTOPILOT_API_KEY;
8const BASE_URL = 'https://api2.autopilothq.com/v1';
9const ONBOARDING_JOURNEY_ID = process.env.AUTOPILOT_ONBOARDING_JOURNEY_ID;
10
11const autopilot = axios.create({
12  baseURL: BASE_URL,
13  headers: {
14    'autopilotapikey': API_KEY,
15    'Content-Type': 'application/json'
16  }
17});
18
19// Add or update a contact (upsert by email)
20app.post('/contacts/register', async (req, res) => {
21  const { email, firstName, lastName, plan, source } = req.body;
22  if (!email) return res.status(400).json({ error: 'email is required' });
23
24  try {
25    const payload = {
26      contact: {
27        Email: email,
28        FirstName: firstName || '',
29        LastName: lastName || '',
30        'string--plan_type': plan || 'free',
31        'string--signup_source': source || 'direct',
32        'integer--signup_timestamp': Math.floor(Date.now() / 1000)
33      }
34    };
35
36    const { data } = await autopilot.post('/contact', payload);
37
38    // Enroll in onboarding journey if configured
39    if (ONBOARDING_JOURNEY_ID) {
40      await autopilot.post(`/journey/${ONBOARDING_JOURNEY_ID}/contact`, {
41        contact: { Email: email }
42      });
43    }
44
45    res.json({ success: true, contactId: data.contact_id });
46  } catch (err) {
47    console.error('Autopilot error:', err.response?.data || err.message);
48    res.status(500).json({ error: err.message });
49  }
50});
51
52// Update a contact property (e.g., plan upgrade event)
53app.post('/contacts/event', async (req, res) => {
54  const { email, event, properties } = req.body;
55  if (!email || !event) return res.status(400).json({ error: 'email and event required' });
56
57  try {
58    // Build contact update with typed custom fields
59    const contactFields = { Email: email };
60    if (properties) {
61      for (const [key, value] of Object.entries(properties)) {
62        const prefix = typeof value === 'number' ? 'integer--' : 'string--';
63        contactFields[`${prefix}${key}`] = value;
64      }
65    }
66    // Add the event name as a custom field for journey triggering
67    contactFields['string--last_event'] = event;
68    contactFields['integer--last_event_timestamp'] = Math.floor(Date.now() / 1000);
69
70    await autopilot.post('/contact', { contact: contactFields });
71    res.json({ success: true });
72  } catch (err) {
73    console.error('Event update error:', err.response?.data || err.message);
74    res.status(500).json({ error: err.message });
75  }
76});
77
78// Unsubscribe a contact from all marketing
79app.post('/contacts/unsubscribe', async (req, res) => {
80  const { email } = req.body;
81  if (!email) return res.status(400).json({ error: 'email is required' });
82
83  try {
84    await autopilot.post('/contact', {
85      contact: { Email: email, unsubscribed: true }
86    });
87    res.json({ success: true });
88  } catch (err) {
89    res.status(500).json({ error: err.message });
90  }
91});
92
93app.listen(3000, '0.0.0.0', () => {
94  console.log('Autopilot integration server running on port 3000');
95});

1from flask import Flask, request, jsonify
2import os
3import json
4
5app = Flask(__name__)
6
7@app.route('/autopilot/webhook', methods=['POST'])
8def autopilot_webhook():
9    try:
10        data = request.get_json(force=True)
11        if not data:
12            return jsonify({'error': 'No JSON body'}), 400
13
14        # Extract contact information from the webhook payload
15        contact = data.get('contact', {})
16        email = contact.get('Email', '')
17        first_name = contact.get('FirstName', '')
18        journey_name = data.get('journey_name', 'unknown')
19        action_type = data.get('type', 'unknown')
20
21        print(f"Autopilot event: {action_type} in journey '{journey_name}' for {email}")
22
23        # Process different event types
24        if action_type == 'journey_completed':
25            # Handle journey completion — e.g., update CRM, notify sales team
26            print(f"Contact {email} completed journey: {journey_name}")
27            # update_crm_lead_score(email, points=50)
28
29        elif action_type == 'email_clicked':
30            link_url = data.get('link_url', '')
31            print(f"Contact {email} clicked: {link_url}")
32            # track_engagement_event(email, 'email_click', link_url)
33
34        elif action_type == 'unsubscribed':
35            print(f"Contact {email} unsubscribed")
36            # mark_user_unsubscribed(email)
37
38        return jsonify({'status': 'received', 'email': email}), 200
39
40    except Exception as e:
41        print(f"Webhook error: {e}")
42        return jsonify({'error': str(e)}), 500
43
44@app.route('/health', methods=['GET'])
45def health():
46    return jsonify({'status': 'ok'}), 200
47
48if __name__ == '__main__':
49    app.run(host='0.0.0.0', port=3000)