How to Integrate Replit with Keap

1import os
2import requests
3from typing import Optional
4
5API_KEY = os.environ["KEAP_API_KEY"]
6BASE_URL = "https://api.infusionsoft.com/crm/rest/v1"
7
8HEADERS = {
9    "Authorization": f"Bearer {API_KEY}",
10    "Content-Type": "application/json"
11}
12
13def find_contact_by_email(email: str) -> Optional[dict]:
14    """Search for an existing contact by email address."""
15    params = {"email": email, "limit": 1}
16    response = requests.get(f"{BASE_URL}/contacts", params=params, headers=HEADERS)
17    response.raise_for_status()
18    contacts = response.json().get("contacts", [])
19    return contacts[0] if contacts else None
20
21def create_contact(email: str, first_name: str, last_name: str = "",
22                  phone: str = "") -> dict:
23    """Create a new contact in Keap."""
24    payload = {
25        "email_addresses": [{"email": email, "field": "EMAIL1"}],
26        "given_name": first_name,
27        "family_name": last_name,
28        "duplicate_option": "Email"  # Merge if email exists
29    }
30    if phone:
31        payload["phone_numbers"] = [{"number": phone, "field": "PHONE1"}]
32
33    response = requests.post(f"{BASE_URL}/contacts", json=payload, headers=HEADERS)
34    response.raise_for_status()
35    return response.json()
36
37def apply_tag(contact_id: int, tag_id: int) -> dict:
38    """Apply a tag to a contact (triggers any automation sequences on that tag)."""
39    payload = {"tagIds": [tag_id]}
40    response = requests.post(
41        f"{BASE_URL}/contacts/{contact_id}/tags",
42        json=payload,
43        headers=HEADERS
44    )
45    response.raise_for_status()
46    return response.json()
47
48def remove_tag(contact_id: int, tag_id: int) -> None:
49    """Remove a tag from a contact."""
50    response = requests.delete(
51        f"{BASE_URL}/contacts/{contact_id}/tags/{tag_id}",
52        headers=HEADERS
53    )
54    response.raise_for_status()
55
56def get_tags() -> list:
57    """Get all tags in the Keap account."""
58    response = requests.get(f"{BASE_URL}/tags", headers=HEADERS)
59    response.raise_for_status()
60    return response.json().get("tags", [])
61
62def create_note(contact_id: int, title: str, body: str,
63                note_type: str = "Other") -> dict:
64    """Add a note to a contact record."""
65    payload = {
66        "contact_id": contact_id,
67        "title": title,
68        "body": body,
69        "type": note_type  # Appointment, Call, Email, Fax, Letter, Meeting, Other
70    }
71    response = requests.post(f"{BASE_URL}/notes", json=payload, headers=HEADERS)
72    response.raise_for_status()
73    return response.json()
74
75# Example usage
76if __name__ == "__main__":
77    # Get all tags to find the right ID
78    tags = get_tags()
79    print("Available tags:")
80    for tag in tags[:10]:
81        print(f"  {tag['id']}: {tag['name']}")
82
83    # Create or find a contact
84    contact = find_contact_by_email("test@example.com")
85    if not contact:
86        contact = create_contact("test@example.com", "Test", "User")
87        print(f"Contact created: ID {contact['id']}")
88    else:
89        print(f"Contact found: ID {contact['id']}")
90
91    # Apply a tag to trigger automation (replace 123 with real tag ID)
92    if tags:
93        apply_tag(contact['id'], tags[0]['id'])
94        print(f"Tag applied: {tags[0]['name']}")

1const express = require('express');
2const axios = require('axios');
3
4const app = express();
5app.use(express.json());
6
7const API_KEY = process.env.KEAP_API_KEY;
8const ONBOARDING_TAG_ID = parseInt(process.env.KEAP_ONBOARDING_TAG_ID || '0');
9const BASE_URL = 'https://api.infusionsoft.com/crm/rest/v1';
10
11const keap = axios.create({
12  baseURL: BASE_URL,
13  headers: {
14    'Authorization': `Bearer ${API_KEY}`,
15    'Content-Type': 'application/json'
16  }
17});
18
19async function findOrCreateContact(email, firstName, lastName) {
20  // Search for existing contact
21  const searchResp = await keap.get('/contacts', { params: { email, limit: 1 } });
22  const contacts = searchResp.data.contacts || [];
23  if (contacts.length > 0) return contacts[0];
24
25  // Create new contact
26  const createResp = await keap.post('/contacts', {
27    email_addresses: [{ email, field: 'EMAIL1' }],
28    given_name: firstName,
29    family_name: lastName || '',
30    duplicate_option: 'Email'
31  });
32  return createResp.data;
33}
34
35// Enroll a new lead and apply onboarding tag
36app.post('/enroll', async (req, res) => {
37  const { email, firstName, lastName, source } = req.body;
38  if (!email || !firstName) {
39    return res.status(400).json({ error: 'email and firstName are required' });
40  }
41
42  try {
43    const contact = await findOrCreateContact(email, firstName, lastName);
44    const contactId = contact.id;
45
46    // Apply onboarding tag to trigger automation sequence
47    if (ONBOARDING_TAG_ID) {
48      await keap.post(`/contacts/${contactId}/tags`, {
49        tagIds: [ONBOARDING_TAG_ID]
50      });
51    }
52
53    // Add a source note
54    if (source) {
55      await keap.post('/notes', {
56        contact_id: contactId,
57        title: 'Lead Source',
58        body: `Contact enrolled from: ${source}`,
59        type: 'Other'
60      });
61    }
62
63    res.json({ success: true, contactId });
64  } catch (err) {
65    console.error('Enroll error:', err.response?.data || err.message);
66    res.status(500).json({ error: err.message });
67  }
68});
69
70// Create an order from an external payment
71app.post('/orders', async (req, res) => {
72  const { email, productName, amount, orderId } = req.body;
73  if (!email || !amount) {
74    return res.status(400).json({ error: 'email and amount are required' });
75  }
76
77  try {
78    // Find the contact
79    const searchResp = await keap.get('/contacts', { params: { email, limit: 1 } });
80    const contacts = searchResp.data.contacts || [];
81    if (!contacts.length) {
82      return res.status(404).json({ error: 'Contact not found in Keap' });
83    }
84
85    const contactId = contacts[0].id;
86
87    // Create the order
88    const orderResp = await keap.post('/orders', {
89      contact_id: contactId,
90      order_date: new Date().toISOString(),
91      order_items: [{
92        description: productName || 'Product Purchase',
93        price: amount,
94        quantity: 1
95      }],
96      title: orderId ? `Order ${orderId}` : 'External Order',
97      source_type: 'ONLINE'
98    });
99
100    res.json({ success: true, orderId: orderResp.data.id });
101  } catch (err) {
102    console.error('Order error:', err.response?.data || err.message);
103    res.status(500).json({ error: err.message });
104  }
105});
106
107app.get('/health', (req, res) => res.json({ status: 'ok' }));
108
109app.listen(3000, '0.0.0.0', () => {
110  console.log('Keap integration server running on port 3000');
111});

1from flask import Flask, request, jsonify
2import os
3import hmac
4import hashlib
5
6app = Flask(__name__)
7HOOK_SECRET = os.environ.get("KEAP_HOOK_SECRET", "")
8
9@app.route('/keap/webhook', methods=['POST'])
10def keap_webhook():
11    # Verify hook secret on first delivery
12    hook_secret = request.headers.get('X-Hook-Secret')
13    if hook_secret:
14        # Keap sends X-Hook-Secret on first delivery for verification
15        return hook_secret, 200, {'X-Hook-Secret': hook_secret}
16
17    data = request.get_json(force=True)
18    if not data:
19        return jsonify({'error': 'No data'}), 400
20
21    event_key = data.get('event_key', 'unknown')
22    object_keys = data.get('object_keys', [])
23
24    print(f"Keap webhook: {event_key}, IDs: {object_keys}")
25
26    if event_key == 'contact.add':
27        for contact_id in object_keys:
28            print(f"New contact added: {contact_id}")
29            # Fetch full contact data if needed
30
31    elif event_key == 'order.add':
32        for order_id in object_keys:
33            print(f"New order created: {order_id}")
34            # Trigger fulfillment workflow
35
36    return jsonify({'status': 'received'}), 200
37
38@app.route('/health', methods=['GET'])
39def health():
40    return jsonify({'status': 'ok'}), 200
41
42if __name__ == '__main__':
43    app.run(host='0.0.0.0', port=3000)