如何用 TwexAPI 获取 XChat 会话列表

在导出 DM 历史之前，先要知道有哪些会话存在。客服系统、CRM 同步或合规归档通常都从 inbox list 开始：conversation ID、一对一还是群组、是否 mute，以及参与者 user ID。

TwexAPI 的 Get Conversations (XChat v3) 端点通过 POST /v3/twitter/conversations 返回 XChat 会话列表。

这个端点只列出会话。它不返回消息历史、媒体文件或消息正文。你可以先用它建立会话清单，再对需要完整消息的会话调用 POST /v3/twitter/dm-history。

端点和请求字段

使用 POST /v3/twitter/conversations 列出当前认证账号的 XChat 会话。

字段	类型	必填	说明
`cookie`	string	是	X 认证 cookie 或 `auth_token`。只保存在服务端。
`count`	integer 或 null	否	每页数量，范围 10 到 200。建议显式设置，避免任务行为不稳定。
`all`	boolean 或 null	否	为 `true` 时会跟随 inbox cursor 返回所有会话。大 inbox 可能较慢。
`proxy`	string 或 null	否	可选 HTTP proxy。

基础请求

curl --request POST \
  --url https://api.twexapi.io/v3/twitter/conversations \
  --header "Authorization: Bearer $TWEXAPI_BEARER_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "cookie": "auth_token=...; ct0=...; twid=...",
    "count": 100,
    "all": false
  }'

响应会在 data 中返回 conversation object 数组。

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "conversation_id": "1928096185664843776:1989842918455291904",
7      "type": "direct",
8      "is_muted": false,
9      "participants": [
10        "1928096185664843776",
11        "1989842918455291904"
12      ]
13    },
14    {
15      "conversation_id": "g1234567890",
16      "type": "group",
17      "is_muted": false,
18      "participants": [
19        "1928096185664843776",
20        "1989842918455291904",
21        "1234567890123456789"
22      ]
23    }
24  ]
25}

把 conversation_id 作为稳定键保存。一对一会话通常类似 a:b，群组会话可能以 g... 开头。

Python 会话同步脚本

这个脚本读取会话列表，并保存规范化行，供后续历史导出任务使用。

1import os
2from datetime import datetime, timezone
3from typing import Any
4
5import requests
6
7API_URL = "https://api.twexapi.io/v3/twitter/conversations"
8BEARER_TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
9X_COOKIE = os.environ["TWEXAPI_X_COOKIE"]
10PROXY = os.environ.get("TWEXAPI_PROXY")
11
12def fetch_conversations(
13    *,
14    count: int = 100,
15    all_conversations: bool = False,
16) -> list[dict[str, Any]]:
17    if count < 10 or count > 200:
18        raise ValueError("count must be between 10 and 200")
19
20    payload: dict[str, Any] = {
21        "cookie": X_COOKIE,
22        "count": count,
23        "all": all_conversations,
24    }
25    if PROXY:
26        payload["proxy"] = PROXY
27
28    response = requests.post(
29        API_URL,
30        headers={
31            "Authorization": f"Bearer {BEARER_TOKEN}",
32            "Content-Type": "application/json",
33            "Accept": "application/json",
34        },
35        json=payload,
36        timeout=90,
37    )
38    response.raise_for_status()
39    data = response.json()
40
41    if data.get("code") != 200:
42        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
43
44    return data["data"]
45
46def normalize_conversation(
47    account_key: str,
48    conversation: dict[str, Any],
49) -> dict[str, Any]:
50    participants = conversation.get("participants") or []
51    return {
52        "account_key": account_key,
53        "conversation_id": conversation["conversation_id"],
54        "conversation_type": conversation["type"],
55        "is_muted": conversation["is_muted"],
56        "participants": participants,
57        "participant_count": len(participants),
58        "synced_at": datetime.now(timezone.utc).isoformat(),
59        "raw": conversation,
60    }
61
62def conversations_for_history_export(
63    rows: list[dict[str, Any]],
64) -> list[dict[str, Any]]:
65    return [
66        row
67        for row in rows
68        if not row["is_muted"] and row["conversation_type"] in {"direct", "group"}
69    ]
70
71if __name__ == "__main__":
72    account_key = "support_account"
73    conversations = fetch_conversations(count=100, all_conversations=True)
74    rows = [normalize_conversation(account_key, item) for item in conversations]
75    export_queue = conversations_for_history_export(rows)
76
77    print(f"Synced {len(rows)} conversations")
78    print(f"Queued {len(export_queue)} conversations for history checks")

运行脚本前，先在服务端环境设置 TWEXAPI_BEARER_TOKEN 和 TWEXAPI_X_COOKIE。不要在浏览器代码里调用这个端点，因为请求需要 X cookie。

什么时候用 all=true

当你需要完整 inbox 清单时使用 all=true，例如首次 CRM 同步或定期审计。API 会在内部跟随 inbox cursor，并返回能拿到的所有会话。

如果只是轻量健康检查或 UI 预览，保持 all: false，并设置有边界的 count。由于响应没有暴露手动 cursor 字段，不要用 conversation_id 自己拼一个分页 cursor。

Conversation 到 History 的工作流

列出会话：调用 POST /v3/twitter/conversations，保存每个 conversation_id。
分类：区分 direct 和 group，并保留参与者 user ID。
选择：根据账号、类型、muted 状态或内部规则，决定哪些会话需要导出消息。
导出历史：对选中的行调用 POST /v3/twitter/dm-history，把 conversation_id 作为 recipient。
解析媒体：如果历史消息包含附件，再对具体媒体调用 POST /v3/twitter/dm-media。

这样每个端点职责清晰：conversations 发现 inbox，history 读取消息，media 解析文件。

应该保存哪些字段

字段	作用
`account_key`	标识这个 inbox 属于哪个 X 账号。
`conversation_id`	后续导出历史的稳定键。
`conversation_type`	区分 `direct` 和 `group` 工作流。
`is_muted`	可用于优先级或排除规则。
`participants`	参与者 user ID，用于匹配 CRM 或客服记录。
`participant_count`	简单检查 direct/group 是否符合预期。
`synced_at`	说明这个 inbox 快照的采集时间。
原始 conversation JSON	字段映射变化时可重新处理。

不要把完整账号 cookie 存在 conversation 行里。保存凭据引用即可。

常见错误

期待这个端点返回消息正文；它只返回 conversation objects。
在大 inbox 的低延迟请求路径里直接使用 all=true。
把 participants 当成展示名；它们是 user ID。
因为群组 conversation ID 不像 a:b，就把 group conversation 丢掉。
响应没有暴露手动 cursor，却用 conversation_id 自己构造假 cursor。

总结

使用 POST /v3/twitter/conversations 建立 XChat inbox 清单：conversation_id、type、is_muted 和 participants。然后用这些 conversation IDs 决定哪些会话继续走 dm-history，需要媒体时再接 dm-media。