如何建立Rocketly与Follow Up Boss（Fub）的连接并完成API集成？

1. 准备工作 #

• 从 Rocketly 后台获取你的 API密钥，通常在「设置-API集成」类菜单下生成
• 从 Follow Up Boss 后台获取 API密钥，路径一般是「设置-开发人员-API密钥」
• 确认两个平台核心API端点：
  • Rocketly：重点关注线索/联系人的读取、创建接口（如GET /api/v1/contacts、POST /api/v1/contacts）
  • Follow Up Boss：重点关注联系人的创建、更新接口（如POST /api/v1/contacts）

2. 基础API连接测试（Python示例） #

先验证两个平台的API可正常调用：

Rocketly API 连接测试 #

import requests

ROCKETLY_API_KEY = "your_rocketly_api_key_here"
ROCKETLY_BASE_URL = "https://api.rocketly.com/v1"

def test_rocketly_connection():
    headers = {
        "Authorization": f"Bearer {ROCKETLY_API_KEY}",
        "Content-Type": "application/json"
    }
    response = requests.get(f"{ROCKETLY_BASE_URL}/contacts", headers=headers)
    if response.status_code == 200:
        print("Rocketly API 连接成功")
        return response.json()
    else:
        print(f"Rocketly API 连接失败: {response.status_code} - {response.text}")

# 执行测试
test_rocketly_connection()

Follow Up Boss API 连接测试 #

import requests

FUB_API_KEY = "your_follow_up_boss_api_key_here"
FUB_BASE_URL = "https://api.followupboss.com/v1"

def test_fub_connection():
    headers = {
        "Authorization": f"Token {FUB_API_KEY}",
        "Content-Type": "application/json"
    }
    response = requests.get(f"{FUB_BASE_URL}/contacts", params={"limit": 1}, headers=headers)
    if response.status_code == 200:
        print("Follow Up Boss API 连接成功")
        return response.json()
    else:
        print(f"Follow Up Boss API 连接失败: {response.status_code} - {response.text}")

# 执行测试
test_fub_connection()

3. 核心集成逻辑示例：将Rocketly线索同步到Follow Up Boss #

以下示例实现从Rocketly获取最新线索并同步到Follow Up Boss：

import requests

# 配置API密钥
ROCKETLY_API_KEY = "your_rocketly_api_key_here"
FUB_API_KEY = "your_follow_up_boss_api_key_here"

# 平台基础URL
ROCKETLY_BASE = "https://api.rocketly.com/v1"
FUB_BASE = "https://api.followupboss.com/v1"

def get_rocketly_new_contacts():
    """获取Rocketly中最近创建的联系人"""
    headers = {"Authorization": f"Bearer {ROCKETLY_API_KEY}"}
    # 可根据业务需求添加时间过滤参数，比如获取24小时内的新线索
    params = {"createdAfter": "2024-01-01T00:00:00Z"}
    response = requests.get(f"{ROCKETLY_BASE}/contacts", headers=headers, params=params)
    if response.status_code == 200:
        return response.json().get("data", [])
    else:
        print(f"获取Rocketly联系人失败: {response.text}")
        return []

def create_fub_contact(contact_data):
    """在Follow Up Boss中创建联系人"""
    headers = {
        "Authorization": f"Token {FUB_API_KEY}",
        "Content-Type": "application/json"
    }
    # 映射Rocketly字段到Follow Up Boss字段，可根据业务调整
    fub_payload = {
        "firstName": contact_data.get("first_name"),
        "lastName": contact_data.get("last_name"),
        "emails": [{"value": contact_data.get("email"), "type": "work"}],
        "phones": [{"value": contact_data.get("phone"), "type": "mobile"}],
        "source": "Rocketly",
        "tags": ["Rocketly线索"]
    }
    response = requests.post(f"{FUB_BASE}/contacts", json=fub_payload, headers=headers)
    if response.status_code in [200, 201]:
        print(f"成功创建FUB联系人: {contact_data.get('email')}")
        return response.json()
    else:
        print(f"创建FUB联系人失败: {response.text}")
        return None

def sync_rocketly_to_fub():
    """同步主逻辑"""
    rocketly_contacts = get_rocketly_new_contacts()
    for contact in rocketly_contacts:
        create_fub_contact(contact)

# 执行同步
sync_rocketly_to_fub()

4. 关键注意事项 #

• 字段映射：根据业务需求调整字段映射逻辑，比如补充地址、公司信息等字段的同步
• 错误处理：生产环境需添加重试机制（如使用tenacity库），处理API调用失败的场景
• Webhook替代轮询：若两个平台支持Webhook，优先用Webhook触发同步，比定时轮询更高效（如Rocketly有新线索时直接推送请求到你的服务，再同步到FUB）
• 权限验证：确保API密钥拥有足够权限（如Rocketly的读取权限、Follow Up Boss的写入权限）

内容的提问来源于stack exchange，提问作者Jibran Ali