Twitter Notifications API, Mentions & Activity Feed

curl -G https://xquik.com/api/v1/x/notifications \
  --data-urlencode "type=Mentions" \
  -H "x-api-key: xq_YOUR_KEY_HERE" | jq

# Page 2
curl -G https://xquik.com/api/v1/x/notifications \
  --data-urlencode "type=Mentions" \
  --data-urlencode "cursor=abc123" \
  -H "x-api-key: xq_YOUR_KEY_HERE" | jq

function notificationsUrl({ cursor, type = "Mentions" }) {
  const params = new URLSearchParams({ type });
  if (cursor) params.set("cursor", cursor);
  return `https://xquik.com/api/v1/x/notifications?${params}`;
}

function toNotificationRows(page, { inboxType }) {
  return page.notifications.map((notification) => ({
    record_type: "notification",
    inbox_type: inboxType,
    notification_id: notification.id,
    notification_type: notification.type ?? null,
    message_preview: notification.message ?? null,
    created_at: notification.timestamp ?? null,
    source_endpoint: "GET /api/v1/x/notifications",
    page_next_cursor: page.has_next_page ? page.next_cursor : null,
  }));
}

async function saveNotificationRows(rows) {
  // Replace this with a private support inbox, CRM, queue, or agent memory write.
  return rows.length;
}

const inboxType = "Mentions";
const response = await fetch(notificationsUrl({ type: inboxType }), {
  headers: { "x-api-key": "xq_YOUR_KEY_HERE" },
});
const data = await response.json();
const rows = toNotificationRows(data, { inboxType });
await saveNotificationRows(rows);

// Paginate
if (data.has_next_page) {
  const next = await fetch(notificationsUrl({ type: inboxType, cursor: data.next_cursor }), {
    headers: { "x-api-key": "xq_YOUR_KEY_HERE" },
  });
  const nextRows = toNotificationRows(await next.json(), { inboxType });
  await saveNotificationRows(nextRows);
}

import requests

def to_notification_rows(page, inbox_type):
    return [
        {
            "record_type": "notification",
            "inbox_type": inbox_type,
            "notification_id": notification["id"],
            "notification_type": notification.get("type"),
            "message_preview": notification.get("message"),
            "created_at": notification.get("timestamp"),
            "source_endpoint": "GET /api/v1/x/notifications",
            "page_next_cursor": page["next_cursor"] if page["has_next_page"] else None,
        }
        for notification in page["notifications"]
    ]

def save_notification_rows(rows):
    # Replace this with a private support inbox, CRM, queue, or agent memory write.
    return len(rows)

inbox_type = "Mentions"
response = requests.get(
    "https://xquik.com/api/v1/x/notifications",
    params={"type": inbox_type},
    headers={"x-api-key": "xq_YOUR_KEY_HERE"},
)
data = response.json()
save_notification_rows(to_notification_rows(data, inbox_type))

# Paginate
while data["has_next_page"]:
    data = requests.get(
        "https://xquik.com/api/v1/x/notifications",
        params={"type": inbox_type, "cursor": data["next_cursor"]},
        headers={"x-api-key": "xq_YOUR_KEY_HERE"},
    ).json()
    save_notification_rows(to_notification_rows(data, inbox_type))