eBay Message API：sendMessageでバイヤーへの自動返信ボットを作る

前回の記事はこちら

【連載#22】eBay Message API：sendMessageでバイヤーへの自動返信ボットを作る

はじめに

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

前回（#21）は、getConversations と getConversation を使い、バイヤーからの受信メッセージを一覧取得・詳細取得する方法を解説しました。会話データの「読み込み」基盤が整ったところで、今回はその直接の続編として、POST /send_message エンドポイントを使った「バイヤーへの返信送信」を実装します。

eBay の販売規模が拡大してくると、「発送はいつですか？」「返品できますか？」「在庫はまだありますか？」といった定型的な問い合わせへの個別対応が、人的リソースを大きく圧迫し始めます。特に日本と海外バイヤーのタイムゾーン差によって営業時間外に届くメッセージへの対応は、購買意欲の冷え込みやネガティブフィードバックの温床になります。本記事では、このような課題をコードで解決する自動返信ボットを段階的に構築します。

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

「自動返信って、なんでも同じ文面を送ればいいんでしょ？　お問い合わせありがとうございます。担当者が確認し、24時間以内にご返信します、とだけ送っておけば十分では？」

これは多くの初学者が最初に抱く疑問であり、よく見られる実装パターンでもあります。確かに技術的には実現可能ですし、Response Rate（返信率）という指標だけを見れば数字は改善されます。しかし eBay のバイヤーエクスペリエンスという観点から見ると、この「一律返信」戦略は長期的に大きなリスクを孕んでいます。

例えば、バイヤーが「追跡番号を教えてください」と具体的に聞いているのに「担当者が確認します」とだけ返信された場合、バイヤーは「このセラーはボットで適当に返信しているだけだ」と感じ、購買後の不安が解消されません。一方、「ご購入ありがとうございます！商品は2営業日以内に発送し、追跡番号は eBay システム経由でお知らせします」という具体的な回答であれば、バイヤーの心理的な安心感は大きく向上します。つまり自動返信の「内容の質」こそが重要なのです。

自動返信ボットを設計する上で最も重要なのは、「何を自動化し、何を人間に任せるか」の境界線を明確に引くことです。発送予定や一般的な商品質問への回答は自動化に向いていますが、返品・クレーム・商品の欠陥報告などのネガティブな内容は、不適切な自動返信によって状況を悪化させる深刻なリスクがあります。これらは必ず人間がエスカレーション対応すべきカテゴリです。

eBay のセラーパフォーマンス評価では、メッセージへの返信速度（Response Rate）が重要指標のひとつです。適切に設計された自動返信ボットは Response Rate を向上させ、Top Rated Seller ステータスの維持にも直接貢献します。しかし「とにかく何か返信すればいい」という思想で作られたボットは、かえってバイヤーの信頼を損ない逆効果になる点を、最初に認識しておく必要があります。

基本的な使い方（ベースライン）：sendMessageで返信を送る

まず最小限の動作確認ができるシンプルな実装から始めましょう。第21回で取得した conversationId を指定して、既存の会話にテキストメッセージを返信する基本形です。

# message_send_baseline.py
import requests
from typing import Optional

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


def send_message_to_buyer(
    access_token: str,
    conversation_id: str,
    message_text: str,
    email_copy_to_sender: bool = False,
    reference_id: Optional[str] = None,
) -> dict:
    """
    既存の会話にバイヤーへの返信を送信する最小実装。

    Args:
        access_token         : OAuth 2.0 アクセストークン
        conversation_id      : 返信先の会話 ID（getConversations で取得）
        message_text         : 送信するメッセージ本文（最大 2000 文字）
        email_copy_to_sender : True の場合、送信者にもメールコピーを送付
        reference_id         : 関連商品の ItemID（任意）

    Returns:
        eBay API のレスポンス dict（成功時は空 dict の場合もある）
    """
    url = f"{EBAY_API_BASE}/send_message"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type":  "application/json",
        "X-EBAY-C-MARKETPLACE-ID": "EBAY_JP",
    }
    payload: dict = {
        "conversationId": conversation_id,
        "messageText":    message_text,
        "emailCopyToSender": email_copy_to_sender,
    }
    # 関連商品を紐付ける場合は reference ブロックを追加
    # ※ referenceType は現在 LISTING のみサポート
    if reference_id:
        payload["reference"] = {
            "referenceId":   reference_id,
            "referenceType": "LISTING",
        }

    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()
    # 成功時はレスポンスボディが空の場合があるため安全に処理する
    return response.json() if response.text.strip() else {}


# ===== 使用例 =====
if __name__ == "__main__":
    TOKEN   = "v^1.1#i^1#f^0#..."          # OAuth アクセストークン
    CONV_ID = "v1|conv_1234567890|0"         # #21 で取得した conversationId

    result = send_message_to_buyer(
        access_token=TOKEN,
        conversation_id=CONV_ID,
        message_text="ご連絡ありがとうございます。商品は現在発送準備中です。",
        email_copy_to_sender=False,
    )
    print("返信送信完了:", result)

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

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

1. キーワード分類の誤判定で見当違いの自動返信を送ってしまうリスク

自動返信ボットを実装する際に最初に直面するのが「分類の精度」問題です。例えば「返品」というキーワードをシンプルな文字列マッチで検出しようとすると、「返品は必要ありません、とても満足しています！」というポジティブなフィードバックメッセージも誤って「返品リクエスト」として分類してしまう可能性があります。

もう一つのよくあるパターンは、複数のカテゴリに跨るメッセージです。「発送はいつですか？もし遅れるようなら返品を検討したいのですが」というメッセージは、発送カテゴリにも返品カテゴリにも該当します。このような複合的なメッセージに発送テンプレートだけを返信するのは状況をさらに悪化させる可能性があります。

実務的な対策としては、キーワードの「優先順位」と「ネガティブパターンの先行評価」を設計することです。具体的には、返品・クレームなどのネガティブカテゴリを最高優先度で判定し、いずれかのネガティブキーワードがヒットした場合は自動返信せずに人間エスカレーションフラグを立てる仕組みにします。評価優先順位は NEGATIVE > RETURN > SHIPPING > QUESTION > UNKNOWN の順に設定することを推奨します。

2. 同じ会話に対する二重送信防止（送信済みフラグ管理）

定期実行（例：5分ごとのバッチ処理）で自動返信ボットを動かす場合、あるサイクルで「未返信」として検出・返信した会話が、次のサイクルでも「未返信」として再度検出される問題が発生します。eBay の API 側でステータスが反映されるまでに数十秒〜数分のタイムラグがあることや、getConversations の conversation_status フィルタが想定通りに動作しないエッジケースが原因です。

この問題を防ぐ最もシンプルな実装は、送信済みの conversationId を Python の Set で管理することです。ただし、プロセス再起動後もデータが失われないよう、本番環境では Redis（SADD/SISMEMBER コマンド）や PostgreSQL（sent_messages テーブル）などの永続化ストレージに送信済み ID を保存することを強く推奨します。

また、getConversations API のレスポンスに含まれる conversation_status フィールドも積極的に活用してください。第21回で解説した通り、ANSWERED ステータスの会話はフィルタリングの段階で除外できます。ただし、前述のタイムラグ問題があるため、API フィルタとアプリ側の重複チェックの両方を二重に持つ「多層防御」の設計が最善です。

3. messageTextの文字数制限を超えた場合のtruncate処理

日本語テンプレートは英語と比べて文字数が少なく見えても、バイヤー名・商品名・配送日数などの動的な値を埋め込んでフォーマットした後に 2000 文字を超えることがあります。特に複数の注文詳細を含む長文テンプレートを使う場合は注意が必要です。len() による文字数チェックを怠ると、本番環境で突然 API エラーが返ってきます。

truncate 処理では単純に先頭 2000 文字で切り捨てると文章の途中で切れてしまい、不自然なメッセージがバイヤーに届きます。句読点（。！？）の位置を rfind() で検索し、2000 文字以内の最後の文区切り位置で切り詰めるロジックを実装することを推奨します。切り詰めが発生した場合は必ずログに WARNING レベルで記録し、テンプレートの長さを見直す契機にしてください。

頻出エラーコード早見表

エラーコード	エラー内容	発生ケース	対処法
13007	MESSAGE_TEXT_TOO_LONG	messageText が 2000 文字を超えている	truncate() で事前に切り詰める。ログに WARNING を記録しテンプレートを見直す
2004	FIELD_VALUE_INVALID	conversationId と otherPartyUsername を同時に指定	返信時は conversationId のみ、新規起票時は otherPartyUsername のみを指定する
6001	RESOURCE_NOT_FOUND	存在しない・アクセス権のない conversationId、または HTTP の mediaUrl を指定	getConversations で最新の ID を再取得する。mediaUrl は必ず HTTPS にする
1100	INVALID_ACCESS_TOKEN	アクセストークンの期限切れ、または message スコープ不足	トークンをリフレッシュし、OAuth スコープに https://api.ebay.com/oauth/api_scope/message が含まれているか確認する

堅牢な実装：getConversations連携と自動返信ボットの完全実装

ここまで解説したすべての実務ポイント（キーワード優先度分類・二重送信防止・文字数制限ハンドリング・営業時間外判定・エスカレーション）を組み込んだ、本番稼働可能な完全実装を示します。

# auto_reply_bot.py
import requests
import re
import logging
from datetime import datetime, time
from zoneinfo import ZoneInfo
from typing import Optional, Dict, List, Set, Tuple, Callable
from dataclasses import dataclass, field
from enum import Enum, auto

logger = logging.getLogger(__name__)

EBAY_API_BASE   = "https://api.ebay.com/commerce/message/v1"
MAX_MESSAGE_LEN = 2000
JST             = ZoneInfo("Asia/Tokyo")


class MessageCategory(Enum):
    NEGATIVE = auto()  # クレーム・詐欺疑惑等（最高優先度でエスカレーション）
    RETURN   = auto()  # 返品・返金・キャンセル（人間対応必須）
    SHIPPING = auto()  # 発送・追跡番号に関する問い合わせ
    QUESTION = auto()  # 在庫・商品スペック等の一般質問
    UNKNOWN  = auto()  # 分類不能（自動返信スキップ）


# キーワードルール（優先度降順で定義すること）
KEYWORD_RULES: List[Tuple[MessageCategory, List[str]]] = [
    (MessageCategory.NEGATIVE, [
        r"詐欺", r"偵物", r"クレーム", r"苦情", r"最悪", r"ひどい",
        r"fraud", r"fake", r"scam", r"complaint",
    ]),
    (MessageCategory.RETURN, [
        r"返品", r"返金", r"refund", r"return", r"キャンセル", r"cancel",
    ]),
    (MessageCategory.SHIPPING, [
        r"発送", r"出荷", r"いつ届", r"配送", r"追跡", r"tracking",
        r"shipping", r"dispatch",
    ]),
    (MessageCategory.QUESTION, [
        r"在庫", r"サイズ", r"カラー", r"色", r"状態", r"コンディション",
        r"stock", r"size", r"color", r"condition",
    ]),
]


REPLY_TEMPLATES: Dict[MessageCategory, str] = {
    MessageCategory.SHIPPING: (
        "ご購入・お問い合わせありがとうございます。\n"
        "ご注文の商品は {shipping_days} 営業日以内に発送いたします。\n"
        "追跡番号は発送完了後、eBay システムを通じて自動でお知らせします。\n"
        "ご不明な点がございましたらお気軽にご連絡ください。"
    ),
    MessageCategory.QUESTION: (
        "お問い合わせありがとうございます。\n"
        "ご質問内容を確認し、担当者より改めてご回答申し上げます。\n"
        "通常 24 時間以内にご返信いたします。もうしばらくお待ちください。"
    ),
}


@dataclass
class AutoReplyBot:
    """
    eBay Message API を使ったバイヤー自動返信ボット。

    Attributes:
        access_token          : OAuth 2.0 アクセストークン
        marketplace_id        : eBay マーケットプレイス ID（例: EBAY_JP）
        business_start        : 営業開始時刻（JST）
        business_end          : 営業終了時刻（JST）
        shipping_days         : 発送予定日数（テンプレートに埋め込む）
        sent_conversation_ids : 送信済み conversationId の Set（二重送信防止）
    """
    access_token: str
    marketplace_id: str  = "EBAY_JP"
    business_start: time = field(default_factory=lambda: time(9, 0))
    business_end: time   = field(default_factory=lambda: time(18, 0))
    shipping_days: int   = 2
    sent_conversation_ids: Set[str] = field(default_factory=set)

    def _headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type":  "application/json",
            "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id,
        }

    # ----------------------------------------------------------
    # 1. 未返信会話の取得（第21回 getConversations を活用）
    # ----------------------------------------------------------
    def get_unanswered_conversations(self, limit: int = 50) -> List[Dict]:
        """UNANSWERED ステータスの会話を最大 limit 件取得する。"""
        url    = f"{EBAY_API_BASE}/get_conversations"
        params = {"conversation_status": "UNANSWERED", "limit": str(limit)}
        resp   = requests.get(url, headers=self._headers(), params=params, timeout=10)
        resp.raise_for_status()
        return resp.json().get("conversations", [])

    # ----------------------------------------------------------
    # 2. キーワード分類エンジン（優先度降順で評価）
    # ----------------------------------------------------------
    def classify(self, text: str) -> MessageCategory:
        """メッセージ本文をキーワードで分類する（優先度順）。"""
        for category, patterns in KEYWORD_RULES:
            for pat in patterns:
                if re.search(pat, text, re.IGNORECASE):
                    return category
        return MessageCategory.UNKNOWN

    # ----------------------------------------------------------
    # 3. 文字数制限チェックと truncate（句読点で区切る）
    # ----------------------------------------------------------
    def _truncate(self, text: str) -> str:
        """2000 文字を超える場合、文の区切りで切り詰める。"""
        if len(text) <= MAX_MESSAGE_LEN:
            return text
        logger.warning("メッセージが %d 文字。%d 文字に切り詰めます。",
                        len(text), MAX_MESSAGE_LEN)
        truncated = text[:MAX_MESSAGE_LEN]
        for delim in ("。", "！", "？", ".", "!", "?"):
            pos = truncated.rfind(delim)
            if pos > MAX_MESSAGE_LEN // 2:
                return truncated[: pos + 1]
        return truncated[:MAX_MESSAGE_LEN - 3] + "..."

    # ----------------------------------------------------------
    # 4. 営業時間判定（JST）
    # ----------------------------------------------------------
    def _is_business_hours(self) -> bool:
        """現在時刻が JST の営業時間内かを判定する。"""
        now_jst = datetime.now(JST).time()
        return self.business_start <= now_jst <= self.business_end

    # ----------------------------------------------------------
    # 5. メッセージ送信（二重送信防止付き）
    # ----------------------------------------------------------
    def send_message(
        self,
        conversation_id: str,
        message_text: str,
        reference_id: Optional[str] = None,
    ) -> bool:
        """
        conversationId を指定してバイヤーに返信を送る。

        Returns:
            True: 送信成功  /  False: 二重送信スキップまたは送信失敗
        """
        if conversation_id in self.sent_conversation_ids:
            logger.info("二重送信防止: %s は送信済みです。", conversation_id)
            return False

        safe_text = self._truncate(message_text)
        payload: Dict = {
            "conversationId": conversation_id,
            "messageText":    safe_text,
            "emailCopyToSender": False,
        }
        if reference_id:
            payload["reference"] = {
                "referenceId":   reference_id,
                "referenceType": "LISTING",
            }

        url = f"{EBAY_API_BASE}/send_message"
        try:
            resp = requests.post(
                url, headers=self._headers(), json=payload, timeout=10
            )
            resp.raise_for_status()
            self.sent_conversation_ids.add(conversation_id)  # 送信成功後に登録
            logger.info("返信成功: conversation_id=%s", conversation_id)
            return True
        except requests.HTTPError as exc:
            logger.error(
                "sendMessage 失敗: status=%d  body=%s",
                exc.response.status_code, exc.response.text,
            )
            return False

    # ----------------------------------------------------------
    # 6. メインループ（バッチ実行のエントリポイント）
    # ----------------------------------------------------------
    def run(
        self,
        escalate_callback: Optional[Callable[[str, MessageCategory, str], None]] = None,
    ) -> None:
        """
        未返信会話を取得 → 分類 → 自動返信する一連の処理を実行する。

        Args:
            escalate_callback: エスカレーション時に呼ばれる callable。
                               引数: (conversation_id, category, message_text)
        """
        conversations = self.get_unanswered_conversations()
        logger.info("%d 件の未返信会話を処理します。", len(conversations))

        for conv in conversations:
            conv_id: str           = conv.get("conversationId", "")
            messages: List[Dict]   = conv.get("messages", [])
            latest_text: str       = messages[-1].get("text", "") if messages else ""
            item_id: Optional[str] = conv.get("itemId")

            category = self.classify(latest_text)
            logger.debug("conv_id=%s  category=%s", conv_id, category.name)

            # --- エスカレーション判定（NEGATIVE / RETURN は人間対応必須）---
            if category in (MessageCategory.NEGATIVE, MessageCategory.RETURN):
                logger.warning(
                    "エスカレーション: conv_id=%s  category=%s",
                    conv_id, category.name,
                )
                if escalate_callback:
                    escalate_callback(conv_id, category, latest_text)
                continue

            # --- 自動返信テンプレートの選択 ---
            template = REPLY_TEMPLATES.get(category)
            if template is None:
                logger.info("自動返信対象外（UNKNOWN）: conv_id=%s", conv_id)
                continue

            # --- 返信テキスト生成（営業時間外メッセージの付記）---
            reply_text = template.format(shipping_days=self.shipping_days)
            if not self._is_business_hours():
                reply_text += (
                    "\n\n※ 現在は営業時間外（9:00～18:00 JST）のため、"
                    "翌営業日に改めてご確認いたします。"
                )

            # --- 送信実行 ---
            self.send_message(conv_id, reply_text, reference_id=item_id)


# ===== エントリポイント =====
if __name__ == "__main__":
    import os
    logging.basicConfig(level=logging.INFO)

    def slack_escalation(
        conv_id: str, category: MessageCategory, text: str
    ) -> None:
        """Slack への通知（実装は slack_sdk 等を使用）"""
        print(f"[ESCALATION] {category.name}: conv={conv_id[:20]}  msg={text[:50]}")

    bot = AutoReplyBot(
        access_token=os.environ["EBAY_ACCESS_TOKEN"],
        shipping_days=2,
    )
    bot.run(escalate_callback=slack_escalation)

このコードのポイントは、エスカレーション処理を escalate_callback という関数オブジェクトで外部から注入できる設計にしていることです。本番環境では slack_sdk を使った Slack 通知、smtplib を使ったメール送信、あるいは Jira への Issue 自動起票など、チームの運用体制に合わせた実装を引数として渡すことができます。ロジックとI/Oが明確に分離されているため、ユニットテストも容易です。

また、sent_conversation_ids を dataclass のフィールドとして保持することで、同一プロセス実行サイクル内の二重送信を防ぎます。プロセス再起動後も永続化するには、Set の追加・参照部分を Redis（pipeline で SADD + SISMEMBER）または PostgreSQL（sent_messages テーブルへの INSERT ON CONFLICT DO NOTHING）と同期させる拡張が必要です。実運用ではこの永続化層を最初から組み込んでおくことを強く推奨します。

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

レート制限対策とキューイングによる非同期送信

eBay Message API にはレート制限（Rate Limit）が設定されています。大規模セラーが数百〜数千件の未返信会話を一括処理しようとすると、連続した POST /send_message リクエストが短時間に集中し、HTTP 429 Too Many Requests が返り始めます。この状態で単純な for ループを回しているだけでは、多くのメッセージ送信がサイレントに失敗してしまいます。

実務的な対策として、以下の2つのアプローチを組み合わせることを推奨します。

【アプローチ 1：Exponential Backoff（指数バックオフ）による自動リトライ】

429 エラーを受け取った際に即座にリトライせず、待機時間を指数関数的に延ばしながら再試行する方式です。tenacity ライブラリを使うことで、バックオフロジックをデコレータ1行で追加できます。

# rate_limited_send.py
import requests
import logging
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)

logger = logging.getLogger(__name__)


class RateLimitError(Exception):
    """eBay API から HTTP 429 を受け取った際に送出するカスタム例外。"""
    pass


def _check_rate_limit(resp: requests.Response) -> None:
    """HTTP 429 を受け取ったら RateLimitError を送出する。"""
    if resp.status_code == 429:
        retry_after = int(resp.headers.get("Retry-After", 60))
        raise RateLimitError(f"Rate limited. Retry-After: {retry_after}s")
    resp.raise_for_status()


@retry(
    retry=retry_if_exception_type(RateLimitError),
    wait=wait_exponential(multiplier=2, min=5, max=120),  # 5s -> 10s -> 20s -> ...
    stop=stop_after_attempt(5),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def send_with_backoff(
    session: requests.Session,
    url: str,
    headers: dict,
    payload: dict,
) -> dict:
    """指数バックオフ付きの sendMessage 呼び出し（最大 5 回リトライ）。"""
    resp = session.post(url, headers=headers, json=payload, timeout=10)
    _check_rate_limit(resp)
    return resp.json() if resp.text.strip() else {}

【アプローチ 2：キューイングによる非同期送信】

数百件を超える規模になると、同期的なバッチ処理では実行時間そのものが問題になります。Python の asyncio と aiohttp を使った非同期処理、または Redis Queue（RQ）や Celery を使ったタスクキューイングへの移行を検討してください。キューイングのアーキテクチャでは「会話の検出ジョブ（ポーリング）」と「メッセージ送信ジョブ」を分離します。検出ジョブは5分ごとに getConversations を呼んで未返信会話を Redis キューに積み、送信ワーカーはキューからタスクを順に取り出して sendMessage を実行します。この設計により送信処理がピーク時間に集中するのを防ぎ、レート制限に引っかかるリスクを大幅に低減できます。

requests.Session オブジェクトを AutoReplyBot インスタンス内で保持して複数の API 呼び出し間で共有することで、コネクションプールが有効になりリクエストのオーバーヘッドを削減できます。大量処理を行う場合は HTTPAdapter の pool_maxsize パラメータをデフォルトの 10 から 20〜30 程度に引き上げることを検討してください。また、クラウド環境（AWS Lambda 等）でボットを動かす際は、ウォームスタート時に Session を再利用するグローバルインスタンスパターンを採用すると、コールドスタートのオーバーヘッドを回避できます。

まとめ

本記事では、eBay Message API の POST /send_message エンドポイントを使ったバイヤー自動返信ボットを、ベースライン実装から本番稼働レベルまで段階的に構築しました。

• ベースライン: conversationId と messageText を指定した最小構成の sendMessage 呼び出しで既存会話への返信機能を実装し、emailCopyToSender・messageMedia・reference などのオプションパラメータの仕様と使い分けを理解する。
• 深いポイント: キーワード分類の優先度設計（NEGATIVE > RETURN > SHIPPING > QUESTION）、Set による二重送信防止、messageText の 2000 文字制限に対応した句読点ベースの truncate 処理、そしてネガティブ・返品メッセージの人間エスカレーション。これらを組み合わせた dataclass ベースの完全実装コードで、実運用に直結するボットを構築する。
• スケーリング: tenacity による Exponential Backoff で HTTP 429 に堅牢に対応し、大規模処理ではキューイングアーキテクチャ（RQ / Celery）と非同期処理（asyncio + aiohttp）への移行でシステムの安定性とスループットを確保する。

次のステップ

次回（#23）は、Message API シリーズの最終回として bulkUpdateConversation エンドポイントを解説します。個別に会話ステータスを更新するのではなく、一度のリクエストで大量の会話を「既読」「アーカイブ」「スター付き」などに一括変更する方法と、カスタマーサポートのワークフローへの組み込み方を紹介します。お楽しみに！

次の記事はこちら