eBay Message API：bulkUpdateConversationで大量の会話ステータスを一括管理する

前回の記事はこちら

【連載#23】eBay Message API：bulkUpdateConversationで大量の会話ステータスを一括管理する

はじめに

本記事は、全42回にわたる「eBay API 実践ガイド」の第23回です。

第21回（getConversationsとgetConversationでバイヤーとのメッセージを取得する）ではMessage API の「読み取り」基盤を、第22回（sendMessageでバイヤーへの自動返信ボットを作る）では「送信」機能をそれぞれ実装しました。本記事は Message API シリーズ（第21〜23回）の総仕上げとして、会話の「ステータス管理」——すなわち既読化・アーカイブ・削除を一括で行う機能を扱います。

自動返信ボットを本番稼働させると、ほどなく別の問題が浮上してきます。それは「対応済みの会話が受信箱に残り続け、まだ未対応の会話が埋もれてしまう」という問題です。件数が少ないうちは手動で既読にできますが、毎日数百件のバイヤーメッセージを処理するスケールになると、ステータス管理そのものが業務ボトルネックになります。本記事では、この課題を解決する POST /bulk_update_conversation エンドポイントを実務レベルで徹底解説します。

背景・なぜこれが重要か (Motivation)

「1件ずつ updateConversation を呼べばいいのでは？　どうせ Python の for ループで回せば大量件数も処理できるでしょう？」

これは Message API を使い始めたばかりの開発者が最初に抱く疑問です。技術的には確かに可能です。しかし、この「ループで単件 API を連打する」アプローチは、実務スケールではすぐに限界を露わにします。

まず API コール数の問題です。仮に 1 日に 500 件の会話を対応済みにしたい場合、単件 API を使うと 500 回の HTTP リクエストが発生します。eBay Message API には 1 日あたりのコール数に上限が設定されており、単純なループは貴重なレート制限クォータを浪費します。一方 bulkUpdateConversation は 1 リクエストで最大 10 件を処理できるため、同じ 500 件なら 50 回のリクエストで済みます。コール数を 90% 削減できる計算です。

次にレイテンシの問題です。Python の requests ライブラリで同期的に 500 件のリクエストを直列実行すると、1 リクエストあたり 200〜500 ms のレイテンシがあるため、合計で 100 秒〜250 秒の処理時間が必要になります。これはバッチジョブの許容実行時間を大幅に超えてしまいます。バルク API を使えばリクエスト回数が 1/10 に減り、処理時間も劇的に短縮されます。

さらに重要なのが「エラー管理の複雑さ」です。単件ループでは、途中のリクエストがエラーになった場合に「どこまで成功したか」を自前で追跡しなければなりません。bulkUpdateConversation はレスポンスに各 conversationId ごとの updateStatus を返すため、バッチ単位での成功・失敗の把握が構造化された形で得られます。この「レスポンスを見れば partial success のどれが失敗したか即座にわかる」という設計が、堅牢な再試行ロジックの実装を容易にします。

基本的な使い方（ベースライン）：updateConversationとbulkUpdateConversationの使い分け

まず、単件更新の updateConversation と一括更新の bulkUpdateConversation の両方を最小限のコードで確認しましょう。Base URL は getConversations・sendMessage と同様に https://api.ebay.com/commerce/message/v1 です。OAuth 2.0 のアクセストークンを取得済みであることを前提とします（認証フローは第1回を参照してください）。

# message_update_baseline.py
import requests
from typing import Literal

BASE_URL = "https://api.ebay.com/commerce/message/v1"

ConversationType = Literal["FROM_MEMBERS", "FROM_EBAY"]
ConvStatus       = Literal["ACTIVE", "ARCHIVE", "DELETE"]
BulkConvStatus   = Literal["ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"]


# ── 単件更新 ──────────────────────────────────────────────────
def update_conversation(
    access_token: str,
    conversation_id: str,
    conversation_type: ConversationType,
    *,
    conversation_status: ConvStatus | None = None,
    read: bool | None = None,
) -> dict:
    """
    POST /update_conversation で1件の会話ステータスを更新する。

    重要: conversation_status と read を同時に指定すると
    read のみが適用され conversation_status は無視される。
    """
    url = f"{BASE_URL}/update_conversation"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP",
    }
    body: dict = {
        "conversationId":   conversation_id,
        "conversationType": conversation_type,
    }
    if conversation_status is not None:
        body["conversationStatus"] = conversation_status
    if read is not None:
        body["read"] = read

    response = requests.post(url, json=body, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json() if response.content else {}


# ── 一括更新（最大10件）────────────────────────────────────────
def bulk_update_conversation(
    access_token: str,
    conversations: list[dict],
) -> dict:
    """
    POST /bulk_update_conversation で最大10件を一括更新する。

    conversations の各要素は:
      { "conversationId": str,
        "conversationType": "FROM_MEMBERS" | "FROM_EBAY",
        "conversationStatus": "ACTIVE" | "ARCHIVE" | "DELETE" | "READ" | "UNREAD" }
    """
    if len(conversations) > 10:
        raise ValueError(f"bulk APIの上限は10件。実際の件数: {len(conversations)}")

    url = f"{BASE_URL}/bulk_update_conversation"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP",
    }
    body = {"conversations": conversations}

    response = requests.post(url, json=body, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json()


# ── 動作確認 ──────────────────────────────────────────────────
if __name__ == "__main__":
    TOKEN = "v^1.1#i^1#..."  # ← OAuthトークン

    # 単件: 会話を既読にする
    result_single = update_conversation(
        TOKEN,
        conversation_id="c_001",
        conversation_type="FROM_MEMBERS",
        read=True,
    )
    print("単件更新:", result_single)

    # 一括: 3件の会話を一度にアーカイブする
    batch = [
        {"conversationId": "c_101", "conversationType": "FROM_MEMBERS",
         "conversationStatus": "ARCHIVE"},
        {"conversationId": "c_102", "conversationType": "FROM_MEMBERS",
         "conversationStatus": "ARCHIVE"},
        {"conversationId": "c_103", "conversationType": "FROM_MEMBERS",
         "conversationStatus": "ARCHIVE"},
    ]
    result_bulk = bulk_update_conversation(TOKEN, batch)
    for item in result_bulk.get("conversations", []):
        print(f"  [{item['conversationId']}] → {item.get('updateStatus', '?')}")

実務で躓く場面・深いポイント (Core)

ベースライン実装で基本的な動作確認ができたら、次は本番運用で必ず直面する実務上の落とし穴とその解決策を見ていきましょう。

1. bulk API の上限は10件——大量データはチャンク分割が必須

bulkUpdateConversation は 1 リクエストあたり最大 10 件までしか処理できません。11 件以上を一度に送ると API はエラーを返します。実務では対応済み会話が数百件単位で溜まることが珍しくないため、リストを 10 件ずつのチャンク（塊）に分割してから繰り返し API を呼ぶ処理が必須になります。

Python でのチャンク分割は itertools.islice や単純なスライスで実装できますが、本番コードでは以下のようなジェネレータ関数として切り出しておくとテストが書きやすくなります。

# chunk_utils.py
from typing import Generator, TypeVar

T = TypeVar("T")

def chunked(lst: list[T], size: int) -> Generator[list[T], None, None]:
    """リストを指定サイズのチャンクに分割するジェネレータ"""
    for i in range(0, len(lst), size):
        yield lst[i : i + size]


# 使用例: 47件のconversationIdリストを10件ずつ処理する
BULK_API_LIMIT = 10

conv_ids = [f"c_{n:03d}" for n in range(47)]  # 47件のサンプルID

for chunk in chunked(conv_ids, BULK_API_LIMIT):
    print(f"このチャンクで処理する件数: {len(chunk)}")
    # → 10件 × 4回 + 7件 × 1回 = 合計5リクエストで47件を処理

47 件の場合、10 件 × 4 回 ＋ 7 件 × 1 回の計 5 回のリクエストで処理が完了します。単件 API を使った場合の 47 回と比べ、リクエスト数を約 90% 削減できます。大量件数になるほどこの差は顕著になるため、「どんな件数でも必ず bulkUpdateConversation を使う」方針を設計段階から徹底してください。

2. conversationStatus と read の同時指定——片方が黙って無視される罠

この仕様は実際に踏んだときのデバッグが非常に難しい罠です。単件 API（POST /update_conversation）で「会話をアーカイブしつつ既読にもしたい」と考え、conversationStatus=ARCHIVE と read=True を同時にリクエストボディに含めたとします。

# NG: conversationStatus と read を同時指定
body = {
    "conversationId":   "c_001",
    "conversationType": "FROM_MEMBERS",
    "conversationStatus": "ARCHIVE",  # ← この値は無視される！
    "read": True,                     # ← こちらだけが適用される
}
# 結果: 会話は「既読」になるが「アーカイブ」には移動しない
# しかもAPIはエラーを返さないため、バグに気づきにくい

API はエラーを返さず HTTP 200 を返します。レスポンスボディが空（または updateStatus=SUCCESSFUL）であるため、「成功した」と誤認してしまいます。しかし実際には conversationStatus の変更が適用されておらず、会話はアーカイブされないまま受信箱に残り続けます。

正しい実装は「2 回に分けてリクエストを送る」か、「バルク API（bulkUpdateConversation）の conversationStatus=ARCHIVE を使ってアーカイブし、その後 conversationStatus=READ で既読にする」かのどちらかです。

# OK: 排他性を意識した2ステップ実装（単件APIの場合）
# Step 1: 既読にする
update_conversation(TOKEN, "c_001", "FROM_MEMBERS", read=True)

# Step 2: アーカイブする（別リクエスト）
update_conversation(TOKEN, "c_001", "FROM_MEMBERS",
                    conversation_status="ARCHIVE")

# OK: バルクAPIで conversationStatus=READ を使う（一括の場合）
bulk_update_conversation(TOKEN, [
    {"conversationId":   "c_001",
     "conversationType": "FROM_MEMBERS",
     "conversationStatus": "READ"},  # バルクAPIではREADが使える
])

「なぜこのような仕様なのか？」という疑問は自然ですが、eBay の公式仕様書にこの挙動は明記されており、ドキュメントを熟読せずに直感で実装すると必ずはまります。コードレビューのチェックリストに「単件 API で conversationStatus と read を同時指定していないか」の項目を追加しておくことを強くお勧めします。

3. Partial success（一部成功・一部失敗）の検出とハンドリング

bulkUpdateConversation の重要な特性として、「バッチ内の一部が成功し、一部が失敗する」partial success（部分成功）があります。単件 API であれば HTTP ステータスコード（200 / 4xx / 5xx）だけで成否を判断できますが、バルク API では HTTP 200 が返っても個々の conversationId のレスポンス内に失敗情報が埋め込まれている場合があります。

# バルクAPIのレスポンス例（一部失敗のケース）
response_body = {
    "conversations": [
        {
            "conversationId": "c_101",
            "updateStatus": "SUCCESSFUL"
        },
        {
            "conversationId": "c_102",
            "updateStatus": "FAILED",       # ← 失敗
            "errors": [
                {
                    "errorId":  850050,
                    "domain":   "API_MESSAGE",
                    "category": "REQUEST",
                    "message":  "Conversation not found or access denied."
                }
            ]
        },
        {
            "conversationId": "c_103",
            "updateStatus": "SUCCESSFUL"
        }
    ]
}

# ── Partial success の検出ロジック ────────────────────────────
def extract_failed_ids(bulk_response: dict) -> list[str]:
    """バルク更新レスポンスから失敗したconversationIdを抽出する"""
    failed = []
    for item in bulk_response.get("conversations", []):
        if item.get("updateStatus") != "SUCCESSFUL":
            failed.append(item["conversationId"])
            errors = item.get("errors", [])
            for err in errors:
                print(f"  [FAILED] {item['conversationId']} "
                      f"errorId={err.get('errorId')} "
                      f"msg={err.get('message')}")
    return failed

failed_ids = extract_failed_ids(response_body)
print(f"失敗件数: {len(failed_ids)}")  # → 失敗件数: 1

HTTP レスポンスが 200 であっても updateStatus が FAILED の要素が含まれている場合、それは「API 呼び出し自体は成功したが、個別処理が失敗した」ことを意味します。レスポンスの conversations 配列を必ずループして各要素の updateStatus を確認する処理は、バルク API を使う上で絶対に省いてはなりません。

よくある失敗原因として、「conversationId がすでに削除済み」「アクセス権限のない会話 ID」「conversationType の指定ミス」などが挙げられます。失敗した conversationId は再試行キューに積んで後で処理するか、アラートとして担当者に通知するフローを設計してください。

堅牢な実装：ConversationBulkUpdater クラスによる大量ステータス管理

ここまで解説したすべての実務ポイント——10 件チャンク分割・排他性の検証・partial success のハンドリング・失敗分の再試行キュー——を組み込んだ、プロダクションレベルの完全実装を示します。型アノテーション・docstring・入力バリデーション・例外処理をすべて備えたConversationBulkUpdater クラスとして設計します。

# conversation_bulk_updater.py
import logging
import time
from dataclasses import dataclass, field
from typing import Generator, Literal

import requests

logger = logging.getLogger(__name__)

BASE_URL = "https://api.ebay.com/commerce/message/v1"
BULK_LIMIT = 10          # bulkUpdateConversation の上限
MAX_RETRY_COUNT = 3      # 失敗時の最大再試行回数
RETRY_DELAY_SEC = 5.0    # 再試行間隔（秒）

ConversationType = Literal["FROM_MEMBERS", "FROM_EBAY"]
BulkStatus       = Literal["ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"]


@dataclass
class BulkUpdateResult:
    """バルク更新の実行結果を保持するデータクラス"""
    total_requested:  int = 0
    total_successful: int = 0
    total_failed:     int = 0
    failed_ids:       list[str] = field(default_factory=list)
    retry_queue:      list[str] = field(default_factory=list)


class ConversationBulkUpdater:
    """
    eBay bulkUpdateConversation を使い、大量の会話ステータスを
    10 件チャンク分割・Partial success ハンドリング・再試行付きで一括更新するクラス。

    Parameters
    ----------
    access_token:       OAuth 2.0 アクセストークン
    conversation_type:  処理対象の会話タイプ（FROM_MEMBERS または FROM_EBAY）
    target_status:      更新後のステータス（READ / ARCHIVE / ACTIVE 等）
    """

    def __init__(
        self,
        access_token: str,
        conversation_type: ConversationType,
        target_status: BulkStatus,
    ) -> None:
        if not access_token:
            raise ValueError("access_token は空にできません")
        if conversation_type not in ("FROM_MEMBERS", "FROM_EBAY"):
            raise ValueError(f"不正な conversationType: {conversation_type}")
        if target_status not in ("ACTIVE", "ARCHIVE", "DELETE", "READ", "UNREAD"):
            raise ValueError(f"不正な target_status: {target_status}")

        self._token         = access_token
        self._conv_type     = conversation_type
        self._target_status = target_status
        self._session       = requests.Session()
        self._session.headers.update({
            "Authorization":           f"Bearer {self._token}",
            "Content-Type":            "application/json",
            "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP",
        })

    # ── チャンク分割ジェネレータ ────────────────────────────────────
    @staticmethod
    def _chunked(
        lst: list[str], size: int
    ) -> Generator[list[str], None, None]:
        for i in range(0, len(lst), size):
            yield lst[i : i + size]

    # ── 単チャンクのバルク呼び出し ─────────────────────────────────
    def _call_bulk_api(self, chunk: list[str]) -> dict:
        """
        bulkUpdateConversation を 1 チャンク（最大10件）分呼び出す。
        HTTP 4xx / 5xx は requests.HTTPError として raise する。
        """
        body = {
            "conversations": [
                {
                    "conversationId":     cid,
                    "conversationType":   self._conv_type,
                    "conversationStatus": self._target_status,
                }
                for cid in chunk
            ]
        }
        resp = self._session.post(
            f"{BASE_URL}/bulk_update_conversation",
            json=body,
            timeout=30,
        )
        resp.raise_for_status()
        return resp.json()

    # ── partial success の解析 ─────────────────────────────────────
    @staticmethod
    def _parse_partial_success(response: dict) -> tuple[list[str], list[str]]:
        """
        バルク API レスポンスを解析し (successful_ids, failed_ids) を返す。
        """
        successful, failed = [], []
        for item in response.get("conversations", []):
            cid = item.get("conversationId", "")
            if item.get("updateStatus") == "SUCCESSFUL":
                successful.append(cid)
            else:
                failed.append(cid)
                for err in item.get("errors", []):
                    logger.warning(
                        "bulk update failed: conversationId=%s errorId=%s message=%s",
                        cid, err.get("errorId"), err.get("message"),
                    )
        return successful, failed

    # ── メインの一括更新メソッド ───────────────────────────────────
    def update_all(
        self,
        conversation_ids: list[str],
        *,
        chunk_delay_sec: float = 0.5,
    ) -> BulkUpdateResult:
        """
        conversation_ids リスト全件を 10 件チャンクで bulkUpdateConversation に送り、
        各チャンクの partial success を検証し、失敗分を再試行キューに積む。

        Parameters
        ----------
        conversation_ids:  更新対象の conversationId のリスト（件数上限なし）
        chunk_delay_sec:   チャンク間の待機時間（レート制限対策、デフォルト0.5秒）

        Returns
        -------
        BulkUpdateResult: 処理結果の集計
        """
        if not conversation_ids:
            logger.warning("conversation_ids が空です。処理をスキップします。")
            return BulkUpdateResult()

        result = BulkUpdateResult(total_requested=len(conversation_ids))
        chunks = list(self._chunked(conversation_ids, BULK_LIMIT))
        logger.info(
            "bulkUpdateConversation 開始: 合計 %d 件 / %d チャンク",
            result.total_requested, len(chunks),
        )

        for idx, chunk in enumerate(chunks, start=1):
            logger.debug("チャンク %d/%d を処理中 (%d件)", idx, len(chunks), len(chunk))
            try:
                response = self._call_bulk_api(chunk)
                ok_ids, ng_ids = self._parse_partial_success(response)
                result.total_successful += len(ok_ids)
                result.total_failed     += len(ng_ids)
                result.failed_ids.extend(ng_ids)
            except requests.HTTPError as exc:
                # チャンク全体が HTTP エラーになった場合（認証エラー等）
                logger.error(
                    "チャンク %d で HTTP エラー: %s — chunk=%s", idx, exc, chunk
                )
                result.total_failed += len(chunk)
                result.failed_ids.extend(chunk)

            # チャンク間のレート制限対策
            if idx < len(chunks):
                time.sleep(chunk_delay_sec)

        result.retry_queue = result.failed_ids.copy()
        logger.info(
            "bulkUpdateConversation 完了: 成功=%d 失敗=%d",
            result.total_successful, result.total_failed,
        )
        return result

    # ── 再試行メソッド ─────────────────────────────────────────────
    def retry_failed(
        self,
        result: BulkUpdateResult,
        max_retries: int = MAX_RETRY_COUNT,
    ) -> BulkUpdateResult:
        """
        BulkUpdateResult.retry_queue に積まれた失敗分を再試行する。
        Exponential backoff で最大 max_retries 回まで再試行する。
        """
        retry_ids = result.retry_queue.copy()
        attempt = 0
        while retry_ids and attempt < max_retries:
            attempt += 1
            wait = RETRY_DELAY_SEC * (2 ** (attempt - 1))
            logger.info(
                "再試行 %d/%d: 対象=%d件 待機=%gs",
                attempt, max_retries, len(retry_ids), wait
            )
            time.sleep(wait)
            sub_result = self.update_all(retry_ids, chunk_delay_sec=1.0)
            # 成功分を集計に反映
            result.total_successful += sub_result.total_successful
            result.total_failed     -= sub_result.total_successful
            retry_ids = sub_result.failed_ids  # 再度失敗した分のみ残す

        result.retry_queue = retry_ids  # 最終的に残った失敗分
        return result


# ── 使用例 ──────────────────────────────────────────────────────
if __name__ == "__main__":
    import os
    logging.basicConfig(level=logging.INFO)

    TOKEN = os.environ["EBAY_ACCESS_TOKEN"]  # 環境変数から取得

    # 200件の対応済み会話をすべてアーカイブする例
    all_ids = [f"c_{n:04d}" for n in range(200)]  # 実際はDBやAPIから取得

    updater = ConversationBulkUpdater(
        access_token=TOKEN,
        conversation_type="FROM_MEMBERS",
        target_status="ARCHIVE",
    )

    result = updater.update_all(all_ids, chunk_delay_sec=0.5)
    print(f"成功: {result.total_successful} / 失敗: {result.total_failed}")

    if result.retry_queue:
        print(f"再試行キュー: {len(result.retry_queue)} 件")
        result = updater.retry_failed(result, max_retries=3)
        print(f"再試行後の残失敗: {len(result.retry_queue)} 件")

このクラスのポイントは、update_all メソッドが「何件でも受け付ける」一方で、内部では必ず BULK_LIMIT（10）件ずつのチャンクに分割して API を呼び出す点です。呼び出し元は「件数の上限を気にせず conversationId のリストを渡すだけ」でよく、チャンク分割の複雑さがカプセル化されています。

retry_failed メソッドは Exponential backoff（指数バックオフ）を採用しており、1 回目の再試行は 5 秒後、2 回目は 10 秒後、3 回目は 20 秒後に実行されます。eBay サーバー側の一時的な障害や過負荷が原因の失敗は、この方式でほとんどの場合に自動回復します。3 回再試行しても失敗が残った場合は、result.retry_queue に残留する conversationId がログ・アラート・DB への記録などの後続処理に引き渡されます。

パフォーマンス・スケーリング視点 (深度)

定期バッチジョブ化と失敗分の監視ダッシュボード設計

自動返信ボット（第22回）と会話ステータス管理（本記事）の両方が実装できたら、次のステップはこれらを定期実行バッチジョブとして組み合わせることです。典型的なカスタマーサービス自動化パイプラインは次のような構成になります。

• 【フェーズ 1: 新着メッセージの取得（第21回）】
  getConversations を差分同期（デルタ同期）で定期ポーリングし、前回チェック以降の新着 UNREAD 会話を取得する。実行間隔: 5〜10 分（JST 営業時間内）／ 30 分〜1 時間（営業時間外）
• 【フェーズ 2: 自動返信処理（第22回）】
  キーワード分類エンジンで問い合わせ内容を判定し、対応可能なものは sendMessage で自動返信する。ネガティブメッセージ・クレームは担当者エスカレーションキューに追加する。
• 【フェーズ 3: ステータス一括更新（本記事）】
  フェーズ 2 で自動返信が完了した conversationId を ConversationBulkUpdater で一括 ARCHIVE する。失敗した ID は retry_queue に記録し、次回バッチで再試行する。

# nightly_cs_batch.py  ——  夜間バッチジョブの例
import os
import logging
from datetime import datetime, timezone, timedelta

from message_client import EbayMessageClient
from auto_reply_bot import AutoReplyBot
from conversation_bulk_updater import ConversationBulkUpdater, BulkUpdateResult

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("nightly_cs_batch")


def run_cs_pipeline(access_token: str) -> None:
    """
    カスタマーサービス自動化の3フェーズパイプラインを実行する。
    """
    since = datetime.now(timezone.utc) - timedelta(hours=6)  # 過去6時間分

    # ── Phase 1: 新着会話の取得 ──────────────────────────────────
    client = EbayMessageClient(access_token)
    new_convs = client.get_unread_since(since)  # conversationId のリスト
    logger.info("Phase1: 新着会話 %d 件を取得", len(new_convs))

    # ── Phase 2: 自動返信処理 ────────────────────────────────────
    bot = AutoReplyBot(access_token)
    replied_ids = []
    for conv in new_convs:
        success = bot.handle(conv)
        if success:
            replied_ids.append(conv["conversationId"])

    logger.info("Phase2: 自動返信完了 %d 件 / スキップ（要人間対応）%d 件",
                len(replied_ids), len(new_convs) - len(replied_ids))

    # ── Phase 3: 返信済み会話を一括アーカイブ ───────────────────
    if not replied_ids:
        logger.info("Phase3: アーカイブ対象なし。スキップ。")
        return

    updater = ConversationBulkUpdater(
        access_token=access_token,
        conversation_type="FROM_MEMBERS",
        target_status="ARCHIVE",
    )
    result: BulkUpdateResult = updater.update_all(replied_ids)

    if result.retry_queue:
        # 再試行後も失敗した場合はアラートを発行
        logger.error(
            "Phase3: 最終的に %d 件のアーカイブに失敗。"
            "監視ダッシュボードに記録します: %s",
            len(result.retry_queue),
            result.retry_queue[:5],  # 先頭5件のみログ出力
        )
        # TODO: Slack 通知 / DB への失敗ログ書き込み

    logger.info(
        "Phase3 完了: 成功=%d 失敗=%d",
        result.total_successful, result.total_failed,
    )


if __name__ == "__main__":
    TOKEN = os.environ["EBAY_ACCESS_TOKEN"]
    run_cs_pipeline(TOKEN)

このパイプラインを cron（Linux/Mac）や Task Scheduler（Windows）、あるいは GitHub Actions の schedule トリガーで定期実行することで、「新着メッセージの検出 → 自動返信 → 対応済みをアーカイブ」というカスタマーサービスのコアフローが完全に自動化されます。

監視の観点では、result.retry_queue に残留した conversationId をPostgreSQL や BigQuery などのデータストアに書き込み、失敗件数・失敗率・連続失敗 ID などを Grafana や Metabase で可視化する監視ダッシュボードを構築することを推奨します。「毎日 X 件以上失敗が続いている」「特定の conversationId が何度再試行しても失敗する」といったアノマリーの早期発見が、システムの安定運用に直結します。

さらに大規模な運用（数千件 / 日）では、ConversationBulkUpdater の update_all をそのまま同期実行するのではなく、Celery や RQ（Redis Queue）を使ったタスクキューイングアーキテクチャへの移行を検討してください。各チャンク（10 件）を 1 タスクとして非同期にエンキューすることで、ワーカーを水平スケールさせながら API レート制限を守りつつ高スループットな処理が実現できます。

まとめ

本記事では、eBay Message API の会話ステータス一括管理エンドポイント——updateConversation（単件）と bulkUpdateConversation（最大10件一括）——を実務視点で徹底解説しました。

• ベースライン: POST /update_conversation では conversationStatus と read の同時指定を避け、POST /bulk_update_conversation ではリクエストごとに最大 10 件の制限を守る。bulkUpdateConversation の conversationStatus は 5 値（ACTIVE / ARCHIVE / DELETE / READ / UNREAD）をサポートしており、単件 API とは仕様が異なる点に注意する。
• 深いポイント: 大量件数の処理には 10 件チャンク分割が必須。conversationStatus と read の排他的挙動（read のみが適用される）はエラーが返らないため特に注意が必要。バルク API では partial success（部分成功・部分失敗）が発生しうるため、レスポンスの conversations 配列を必ず走査して失敗 ID を再試行キューに積む実装が不可欠。
• スケーリング: ConversationBulkUpdater クラスにチャンク分割・Exponential backoff・再試行キューをカプセル化し、3 フェーズパイプライン（取得 → 返信 → アーカイブ）として定期バッチジョブ化する。大規模運用では Celery / RQ によるタスクキューイングへの移行と、失敗件数の監視ダッシュボード構築が安定運用の鍵となる。

これで Message API シリーズ（第21〜23回）が完結しました。第21回で「読む」、第22回で「送る」、そして本記事で「管理する」——この3 つのエンドポイントを組み合わせることで、eBay のカスタマーサービス業務をエンドツーエンドで自動化する完全なシステムを構築できるようになりました。自動返信ボットと一括ステータス管理を実際の本番環境に繋ぎ込み、毎日の受信箱が常にクリーンな状態に保たれる体験を、ぜひ実感してください。

次のステップ

次回（#24）は、本シリーズにとって大きな転換点となる回です。これまで第1〜23回にわたって取り組んできたTrading API / Message API 系のフェーズを卒業し、いよいよ 【Sell REST API フェーズ】 へと突入します。

第24回のテーマは 「Trading APIからREST APIへ：OAuth移行と最初のInventory API呼び出し」です。eBay が次世代の出品管理基盤として推進する Inventory API（REST）の全体像と、Trading API の AddFixedPriceItem から移行する際の思想的・実装的な違いを解説します。Authentication / Authorization の仕組みが XML + AuthToken からOAuth 2.0 Bearer Token へと標準化されるこの移行は、API 設計の近代化において最も重要なステップのひとつです。Message API シリーズで OAuth 2.0 に慣れ親しんだ皆さんなら、この移行をスムーズに進められるはずです。お楽しみに！

次の記事はこちら