Code Examples

This page provides complete, copy-paste-ready examples for common Synaptiq API operations in curl, JavaScript (fetch), and Python (requests). Replace sk_live_YOUR_API_KEY with your actual API key.

Creating a Lead

curl

curl -X POST https://synaptiqintel.com/api/v1/leads \
  -H "Authorization: Bearer sk_live_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "email": "maria.garcia@cloudsoft.io",
    "name": "Maria Garcia",
    "company": "CloudSoft",
    "title": "Head of Revenue",
    "phone": "+1-555-0234",
    "tags": ["inbound", "saas"],
    "customFields": {
      "industry": "Cloud Infrastructure",
      "teamSize": "100-250"
    }
  }'

JavaScript (fetch)

async function createLead() {
  const response = await fetch("https://synaptiqintel.com/api/v1/leads", {
    method: "POST",
    headers: {
      "Authorization": "Bearer sk_live_YOUR_API_KEY",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      email: "maria.garcia@cloudsoft.io",
      name: "Maria Garcia",
      company: "CloudSoft",
      title: "Head of Revenue",
      phone: "+1-555-0234",
      tags: ["inbound", "saas"],
      customFields: {
        industry: "Cloud Infrastructure",
        teamSize: "100-250",
      },
    }),
  });

  if (!response.ok) {
    const error = await response.json();
    throw new Error(`API error ${response.status}: ${error.error.message}`);
  }

  const lead = await response.json();
  console.log("Lead created:", lead.id);
  return lead;
}

Python (requests)

import requests

API_KEY = "sk_live_YOUR_API_KEY"
BASE_URL = "https://synaptiqintel.com/api/v1"

def create_lead():
    response = requests.post(
        f"{BASE_URL}/leads",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "email": "maria.garcia@cloudsoft.io",
            "name": "Maria Garcia",
            "company": "CloudSoft",
            "title": "Head of Revenue",
            "phone": "+1-555-0234",
            "tags": ["inbound", "saas"],
            "customFields": {
                "industry": "Cloud Infrastructure",
                "teamSize": "100-250",
            },
        },
    )
    response.raise_for_status()
    lead = response.json()
    print(f"Lead created: {lead['id']}")
    return lead

Listing Leads with Filters

curl

curl "https://synaptiqintel.com/api/v1/leads?status=qualified&sortBy=score&sortOrder=desc&limit=10" \
  -H "Authorization: Bearer sk_live_YOUR_API_KEY"

JavaScript (fetch)

async function listQualifiedLeads() {
  const params = new URLSearchParams({
    status: "qualified",
    sortBy: "score",
    sortOrder: "desc",
    limit: "10",
  });

  const response = await fetch(
    `https://synaptiqintel.com/api/v1/leads?${params}`,
    {
      headers: {
        "Authorization": "Bearer sk_live_YOUR_API_KEY",
      },
    }
  );

  if (!response.ok) {
    const error = await response.json();
    throw new Error(`API error ${response.status}: ${error.error.message}`);
  }

  const result = await response.json();
  console.log(`Found ${result.pagination.total} qualified leads`);

  for (const lead of result.data) {
    console.log(`- ${lead.name} (${lead.company}) — Score: ${lead.score}`);
  }

  return result;
}

Python (requests)

def list_qualified_leads():
    response = requests.get(
        f"{BASE_URL}/leads",
        headers={"Authorization": f"Bearer {API_KEY}"},
        params={
            "status": "qualified",
            "sortBy": "score",
            "sortOrder": "desc",
            "limit": 10,
        },
    )
    response.raise_for_status()
    result = response.json()

    print(f"Found {result['pagination']['total']} qualified leads")
    for lead in result["data"]:
        print(f"  - {lead['name']} ({lead['company']}) — Score: {lead['score']}")

    return result

Sending a Chat Message

curl (Non-Streaming)

curl -X POST https://synaptiqintel.com/api/v1/chat \
  -H "Authorization: Bearer sk_live_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "I need a tool to help my sales team qualify leads faster. What do you offer?",
    "visitorId": "visitor_web_9a8b7c6d",
    "metadata": {
      "page": "/solutions/sales",
      "utm_source": "linkedin"
    }
  }'

JavaScript (fetch) — Non-Streaming

async function sendChatMessage(message, visitorId, sessionId = null) {
  const body = {
    message,
    visitorId,
    metadata: { page: window.location.pathname },
  };

  if (sessionId) {
    body.sessionId = sessionId;
  }

  const response = await fetch("https://synaptiqintel.com/api/v1/chat", {
    method: "POST",
    headers: {
      "Authorization": "Bearer sk_live_YOUR_API_KEY",
      "Content-Type": "application/json",
    },
    body: JSON.stringify(body),
  });

  if (!response.ok) {
    const error = await response.json();
    throw new Error(`API error ${response.status}: ${error.error.message}`);
  }

  const reply = await response.json();
  console.log("AI response:", reply.content);
  console.log("Session ID:", reply.sessionId);
  return reply;
}

// Usage: start a new conversation
const reply = await sendChatMessage(
  "What pricing plans do you offer?",
  "visitor_web_9a8b7c6d"
);

// Continue the conversation
const followUp = await sendChatMessage(
  "Tell me more about the Pro plan",
  "visitor_web_9a8b7c6d",
  reply.sessionId
);

Python (requests)

def send_chat_message(message, visitor_id, session_id=None):
    payload = {
        "message": message,
        "visitorId": visitor_id,
        "metadata": {"source": "python_script"},
    }

    if session_id:
        payload["sessionId"] = session_id

    response = requests.post(
        f"{BASE_URL}/chat",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
    )
    response.raise_for_status()
    reply = response.json()

    print(f"AI: {reply['content']}")
    print(f"Intent: {reply['intent']} (confidence: {reply['confidence']})")
    print(f"Session: {reply['sessionId']}")
    return reply

# Start a conversation
reply = send_chat_message(
    "What pricing plans do you offer?",
    "visitor_py_1a2b3c4d"
)

# Continue it
follow_up = send_chat_message(
    "Tell me about the Enterprise plan",
    "visitor_py_1a2b3c4d",
    session_id=reply["sessionId"]
)

Webhook Listener (Node.js + Express)

A complete webhook receiver that verifies signatures and processes events:

const express = require("express");
const crypto = require("crypto");

const app = express();
const WEBHOOK_SECRET = "whsec_your_signing_secret_here";

// Important: use raw body for signature verification
app.use("/webhooks/synaptiq", express.raw({ type: "application/json" }));

app.post("/webhooks/synaptiq", (req, res) => {
  const timestamp = req.headers["x-synaptiq-timestamp"];
  const signature = req.headers["x-synaptiq-signature"];
  const rawBody = req.body.toString("utf-8");

  // 1. Verify timestamp freshness (5-minute window)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > 300) {
    console.error("Webhook rejected: timestamp too old");
    return res.status(400).json({ error: "Timestamp too old" });
  }

  // 2. Verify HMAC signature
  const expected = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(`${timestamp}.${rawBody}`)
    .digest("hex");

  const trusted = Buffer.from(expected, "hex");
  const received = Buffer.from(signature, "hex");

  if (!crypto.timingSafeEqual(trusted, received)) {
    console.error("Webhook rejected: invalid signature");
    return res.status(401).json({ error: "Invalid signature" });
  }

  // 3. Parse and route the event
  const event = JSON.parse(rawBody);
  console.log(`Received event: ${event.type} (${event.id})`);

  switch (event.type) {
    case "lead.created":
      handleLeadCreated(event.data);
      break;

    case "lead.qualified":
      handleLeadQualified(event.data);
      break;

    case "conversation.handoff":
      handleHandoff(event.data);
      break;

    case "intent.high_value":
      handleHighValueIntent(event.data);
      break;

    default:
      console.log(`Unhandled event type: ${event.type}`);
  }

  // 4. Respond immediately with 200
  res.status(200).json({ received: true });
});

function handleLeadCreated(data) {
  const { lead } = data;
  console.log(`New lead: ${lead.name} (${lead.email}) from ${lead.company}`);
  // Sync to your CRM, send Slack notification, etc.
}

function handleLeadQualified(data) {
  const { lead, qualificationReason } = data;
  console.log(`Lead qualified: ${lead.name} — ${qualificationReason}`);
  // Assign to sales rep, create task in project management tool
}

function handleHandoff(data) {
  const { conversation, handoff } = data;
  console.log(`Handoff requested: ${handoff.reason} (${handoff.priority})`);
  // Route to appropriate human agent or queue
}

function handleHighValueIntent(data) {
  console.log(`High-value intent: ${data.intent} — $${data.estimatedDealSize}`);
  // Alert sales team, escalate priority
}

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Webhook listener running on port ${PORT}`);
});

Error Handling Pattern

A reusable wrapper for handling Synaptiq API errors with rate limit awareness:

JavaScript

async function synaptiqFetch(path, options = {}) {
  const url = `https://synaptiqintel.com/api/v1${path}`;
  const response = await fetch(url, {
    ...options,
    headers: {
      "Authorization": "Bearer sk_live_YOUR_API_KEY",
      "Content-Type": "application/json",
      ...options.headers,
    },
  });

  if (response.status === 429) {
    const retryAfter = response.headers.get("X-RateLimit-RetryAfter") || 5;
    console.log(`Rate limited. Retry after ${retryAfter}s`);
    throw new Error(`Rate limited. Retry after ${retryAfter} seconds.`);
  }

  if (!response.ok) {
    const body = await response.json();
    throw new Error(`[${response.status}] ${body.error.code}: ${body.error.message}`);
  }

  return response.json();
}

Python

from requests.exceptions import HTTPError

def synaptiq_request(method, path, **kwargs):
    url = f"{BASE_URL}{path}"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = requests.request(method, url, headers=headers, **kwargs)

    try:
        response.raise_for_status()
    except HTTPError:
        error_body = response.json().get("error", {})
        if response.status_code == 429:
            retry = response.headers.get("X-RateLimit-RetryAfter", 5)
            print(f"Rate limited. Retry after {retry}s")
        else:
            print(f"[{response.status_code}] {error_body.get('code')}: {error_body.get('message')}")
        raise

    return response.json()