偵測新訊息

要讓自己的系統得知對話有新訊息（例如顯示通知、提醒真人客服），本篇說明建議做法：優先用 Webhook 推送；需要輪詢時只讀有更新的對話、每段對話只讀第一頁

想在自己的系統顯示「有新訊息」的提醒、或在客戶留言時通知真人客服，建議依下列順序選擇做法：

1. 優先用 Webhook 推送：對話平台的訊息 Webhook每則訊息推送一次，不需要輪詢（見用 Webhook 取代輪詢）。

2. 只關心轉真人：用轉真人 Webhook 事件，只在轉真人流程的狀態改變時推送。

3. 必須輪詢時：依本篇的輪詢建議做法，只找出有更新的對話、每段對話只讀訊息列表的第一頁。

請不要每輪把整段對話歷史從最新翻到最舊重抓。一段數百則訊息的對話每輪要讀好幾頁，每頁都是一次完整查詢；每隔幾秒對多段對話這樣做，會讓你的輪詢本身變慢，也會拖慢平台的整體回應速度。

用 Webhook 取代輪詢

需求	建議用法	何時推送
客戶每傳一則訊息就通知	對話平台的訊息 Webhook，類型選 Incoming	每則客戶（聯絡人）訊息
同步 AI 或客服送出的回覆	對話平台的訊息 Webhook，類型選 Outgoing	每則送往客戶所在對話平台的回覆
對話需要真人接手時通知	轉真人 Webhook 事件	只在轉真人、指派客服、轉回 AI、排隊逾時時推送，不會每則訊息都推送

設定位置：後台 → 客服對話 → 對話平台 → 選擇對話平台 → Webhook 設定 → 新增 Webhook，類型選 Incoming 或 Outgoing（後台 Incoming 的說明文字雖為「接收外部請求」，實際是把每則客戶訊息推送到你設定的網址；接收外部訊息的是 Receive），Payload 格式選 MaiAgent Standard（MaiAgent 標準格式）。推送內容、來源驗證與重試行為見訊息 Webhook。

WhatsApp 對話平台的設定頁目前沒有 Webhook 設定 分頁，請改用下方的輪詢做法。

兩種 Webhook 是各自獨立的設定，可同時啟用；推送內容都以訊息或事件的 id 去重。訊息 Webhook 每則只推送一次、失敗不會重試，不保證送達。下方的輪詢做法可以低頻補查客戶訊息（Incoming）；AI 或客服的回覆（Outgoing）推送失敗時，輪詢範例不一定會讀到，請改查 Webhook 的傳送紀錄找出失敗的推送（見訊息 Webhook → 回應與重試）。

輪詢建議做法

沒辦法接收 Webhook（例如接收端無法對外開放）時，請用下面兩個 API 組合輪詢。

1. 找出有更新的對話

GET /api/v1/conversations/?inbox=<對話平台 id>&page_size=20

對話列表固定依最後訊息時間由新到舊排列（回應中的 lastMessageCreatedAt）。每段對話記下上次看到的 lastMessageCreatedAt 與最後一則訊息的 lastMessage.id，兩者任一不同就是有更新的對話（時間只精確到秒，同一秒連傳兩則時只有 id 會變）；每輪從第一筆往下讀，讀到比「目前看過的最新時間往前 10 分鐘」還舊的對話就停。一般情況只需要讀第一頁；只有這段時間內更新的對話超過 page_size 段，才需要跟著回應的 next 讀下一頁。

往前多看 10 分鐘是因為：以 API 對話時，對話的 lastMessageCreatedAt 可能在 AI 回覆產生完才寫入，寫入的仍是客戶訊息的時間，所以有可能比你上一輪看過的其他對話還舊。只比較「是否比上輪最新時間新」會漏掉這類對話。

2. 每段有更新的對話只讀第一頁訊息

GET /api/v1/messages/?conversation=<對話 id>&page_size=20

訊息列表的第一頁就是最新的訊息：系統先取最新的 page_size 則，頁內再依時間由舊到新排列，所以最後一則就是整段對話最新的訊息。回應中的 next 指向更舊的訊息，偵測新訊息時不要每輪都往回翻。

以訊息 id 判斷是否處理過：第一頁裡沒處理過的 id 就是新訊息。只有一種情況需要翻頁——第一頁全部都是沒處理過的訊息（兩輪之間進來超過 page_size 則），才跟著 next 往更舊讀，讀到處理過的訊息為止。

3. 輪詢間隔與每頁筆數

設定	建議	說明
輪詢間隔	30–60 秒	需要更即時的通知請改用 Webhook，不要把間隔縮到幾秒
page_size	20	預設 20、上限 100。每頁筆數越大，單次回應越慢、內容越大

• 回應欄位為 camelCase；lastMessageCreatedAt、createdAt 是毫秒單位的時間戳記字串（例如 "1790753081000"），但只精確到秒——同一秒內可能有多則訊息，請以訊息 id 去重，不要只比時間。

• 範例以本機時間作為開始偵測的起點，執行的主機請開啟網路校時（NTP），避免時鐘偏快而略過剛開始時的訊息。

• 對話列表只列出有客戶訊息的對話；只有開場問候、客戶尚未發話的對話不會出現。

• 客戶傳來新訊息時 lastMessageCreatedAt 一定會更新，但可能晚於訊息本身寫入（見上方步驟 1）；AI 自動回覆不一定會更新它。因此本篇的輪詢做法只偵測客戶訊息：AI 回覆只有在讀取當下已產生才會一併讀到；客戶訊息讀過之後才產生的 AI 回覆，若沒有讓對話的最後訊息時間或最後一則訊息改變，之後的輪詢會略過這段對話，不會再讀到這則回覆。若單則 AI 回覆產生超過 10 分鐘（例如長時間執行工具），該則客戶訊息也可能延到對話下一次更新才被偵測。需要即時取得每則客戶訊息與回覆，請使用 Incoming／Outgoing 訊息 Webhook（每則只推送一次、失敗不重試）。

呼叫範例

cURL

# 1. 依最後訊息時間由新到舊列出對話平台的對話
curl "https://api.maiagent.ai/api/v1/conversations/?inbox=YOUR_INBOX_ID&page_size=20" \
  -H "Authorization: Api-Key YOUR_API_KEY"

# 2. 只讀有更新的對話的第一頁訊息（最新 20 則，頁內舊→新）
curl "https://api.maiagent.ai/api/v1/messages/?conversation=CONVERSATION_ID&page_size=20" \
  -H "Authorization: Api-Key YOUR_API_KEY"

Python

import os
import time

import requests

API = "https://api.maiagent.ai/api/v1"
HEADERS = {"Authorization": f"Api-Key {os.environ['MAIAGENT_API_KEY']}"}
INBOX_ID = os.environ["INBOX_ID"]
POLL_SECONDS = 30
PAGE_SIZE = 20
LOOKBACK_MS = 10 * 60 * 1000  # 往前多看 10 分鐘：對話的最後訊息時間可能在 AI 回覆產生後才寫入


def get(url: str, params: dict | None = None) -> dict:
    resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
    resp.raise_for_status()
    return resp.json()


def recent_conversations(since_ms: int) -> list[dict]:
    """對話列表依最後訊息時間由新到舊排列：讀到比 since_ms 舊的就停，通常只讀第一頁。"""
    conversations = []
    url, params = f"{API}/conversations/", {"inbox": INBOX_ID, "page_size": PAGE_SIZE}
    while url:
        page = get(url, params)
        for conv in page["results"]:
            if int(conv["lastMessageCreatedAt"]) < since_ms:
                return conversations
            conversations.append(conv)
        url, params = page["next"], None
    return conversations


def new_messages(conversation_id: str, seen_ids: set[str], not_before_ms: int) -> list[dict]:
    """讀第一頁（最新的訊息，頁內由舊到新）；只有整頁都沒處理過時，才往更舊的頁讀。"""
    found = []
    url, params = f"{API}/messages/", {"conversation": conversation_id, "page_size": PAGE_SIZE}
    while url:
        page = get(url, params)
        fresh = [m for m in page["results"] if m["id"] not in seen_ids and int(m["createdAt"]) >= not_before_ms]
        found = fresh + found
        if len(fresh) < len(page["results"]):
            break  # 這一頁已碰到處理過（或開始偵測前）的訊息，不必再往回翻
        url, params = page["next"], None
    return found


def change_key(conv: dict) -> tuple:
    """時間只精確到秒：同一秒連傳兩則時只有最後一則訊息的 id 會變，所以一起比對。"""
    return conv["lastMessageCreatedAt"], (conv.get("lastMessage") or {}).get("id")


def main() -> None:
    started_ms = int(time.time()) * 1000  # 從現在（取整到秒）開始偵測，不回頭處理歷史訊息
    newest_ms = started_ms
    last_seen: dict[str, tuple] = {}  # 每段對話上次看到的 change_key（正式環境請持久化或定期清理）
    seen_ids: dict[str, set[str]] = {}  # 每段對話已處理過的訊息 id（同上）
    while True:
        try:
            round_newest = newest_ms
            for conv in recent_conversations(newest_ms - LOOKBACK_MS):
                if last_seen.get(conv["id"]) == change_key(conv):
                    continue  # 這段對話沒有新的客戶訊息（之後才產生的 AI 回覆不會在這裡讀到）
                ids = seen_ids.setdefault(conv["id"], set())
                for msg in new_messages(conv["id"], ids, started_ms):
                    print(f"[新訊息] 對話 {conv['id']} {msg['type']}：{msg['content'][:40]}")
                    ids.add(msg["id"])
                last_seen[conv["id"]] = change_key(conv)
                round_newest = max(round_newest, int(conv["lastMessageCreatedAt"]))
            newest_ms = round_newest  # 整輪成功才前進
        except requests.RequestException as exc:
            print(f"本輪查詢失敗，下一輪重試：{exc}")  # 回看起點沒有前進，未處理完的對話下一輪會補上
        time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    main()

API 金鑰的權限與金鑰擁有者在組織中的角色相同，擁有者需能查看該對話平台的對話，詳見 API 認證 → API 金鑰的權限範圍。