SetNotificationPreferencesでeBayイベント通知を設定してPythonサーバーで受信する

前回の記事はこちら

【連載#15】eBay Trading API：SetNotificationPreferencesでeBayイベント通知を設定してPythonサーバーで受信する

はじめに

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

前回（#14）は、GetMemberMessages / AddMemberMessageAAQToPartnerを使い、バイヤーからの問い合わせに自動返信するCS（カスタマーサポート）システムを構築しました。メッセージ管理の自動化により応答時間を劇的に短縮できましたが、「注文が確定したか」「入金が完了したか」を知るためには、依然として定期的なAPIポーリングが必要でした。

本記事では、このポーリング問題を根本から解決します。eBay の Platform Notifications 機能を使い、注文確定・入金・フィードバックなどのビジネスイベントが発生した瞬間に eBay から自分のサーバーへ通知を Push させるイベント駆動アーキテクチャを構築します。Trading API の SetNotificationPreferences で通知先 URL とイベント種別を登録し、FastAPI で受信エンドポイントを実装します。

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

「定期的に GetOrders を叩けば十分じゃないのか？」

eBay 開発を始めたエンジニアの多くが最初にこう考えます。実際、5分おきに GetOrders を実行すれば新規注文を概ね検知できます。しかしこれは「動く」だけであって、「正しいアーキテクチャ」ではありません。

eBay の Trading API には「1日あたりの API 呼び出し数上限（API Call Limit）」があります。GetOrders を5分おきに実行すると1日288回のコールを消費します。複数のセラーアカウントを管理していたり、出品・在庫更新などの API 操作も並行して行う場合、この上限はあっという間に枯渇します。ポーリング間隔を短くするほど消費が速くなるという本質的なジレンマがあり、「もっとリアルタイムに」という要求を満たすほどコスト（API 消費）が跳ね上がります。

一方、Platform Notifications はイベント駆動型（Event-Driven）アーキテクチャです。eBay 側でイベントが発生したタイミングで、あなたのサーバーへ HTTPS POST リクエストが飛んできます。ポーリングのような API 呼び出し消費はゼロです。注文確定から数秒以内に通知が届くため、発送処理や在庫更新などの後続処理を即座に起動できます。

規模が拡大するほどこの差は顕著になります。月間 500 注文のセラーにとってはポーリングでも許容範囲ですが、月間 5,000 注文を超えてくると、通知ベースアーキテクチャは「あると便利」から「絶対に必要」へと変わります。

基本的な使い方（ベースライン）：SetNotificationPreferencesで通知URLを登録する

まず zeep ライブラリを使って SetNotificationPreferences SOAP API を呼び出し、通知 URL と購読したいイベント種別を登録します。その後 GetNotificationPreferences で登録内容を確認する流れも合わせて示します。事前に pip install zeep requests を実行してください。

SetNotificationPreferences には大きく2つの設定ブロックがあります。ApplicationDeliveryPreferences は通知先 URL やペイロード形式といったアプリケーション全体の設定、UserDeliveryPreferenceArray は購読する個別イベント種別の有効/無効リストです。

# set_notification_prefs.py（ベースライン）
from zeep import Client, Settings
from zeep.transports import Transport
import requests

WSDL_URL    = "https://developer.ebay.com/webservices/latest/eBaySvc.wsdl"
API_VERSION = "1311"
SANDBOX_EP  = "https://api.sandbox.ebay.com/ws/api.dll"
PROD_EP     = "https://api.ebay.com/ws/api.dll"
SITE_ID_JP  = "101"          # eBay Japan

def _make_session(config: dict, call_name: str, is_sandbox: bool) -> requests.Session:
    """Trading API 呼び出し用の HTTPヘッダー付き Session を生成する"""
    session = requests.Session()
    ep = SANDBOX_EP if is_sandbox else PROD_EP
    session.headers.update({
        "X-EBAY-API-CALL-NAME"           : call_name,
        "X-EBAY-API-SITEID"              : SITE_ID_JP,
        "X-EBAY-API-COMPATIBILITY-LEVEL" : API_VERSION,
        "X-EBAY-API-APP-NAME"            : config["app_id"],
        "X-EBAY-API-DEV-NAME"            : config["dev_id"],
        "X-EBAY-API-CERT-NAME"           : config["cert_id"],
    })
    return session

def register_notification_url(
    config: dict,
    notification_url: str,
    events: list,
    is_sandbox: bool = False
) -> None:
    """
    SetNotificationPreferences を呼び出し、通知URLとイベント種別を登録する。

    Args:
        config          : app_id / dev_id / cert_id / user_token を含む辞書
        notification_url: eBay からの通知を受け取る HTTPS URL
        events          : 購読するイベント種別リスト
        is_sandbox      : Sandbox 環境の場合は True
    """
    session  = _make_session(config, "SetNotificationPreferences", is_sandbox)
    settings = Settings(strict=False, xml_huge_tree=True)
    client   = Client(wsdl=WSDL_URL, settings=settings,
                      transport=Transport(session=session))

    notification_enables = [
        {"EventType": ev, "EventEnable": "Enable"} for ev in events
    ]

    response = client.service.SetNotificationPreferences(
        RequesterCredentials={"eBayAuthToken": config["user_token"]},
        ApplicationDeliveryPreferences={
            "ApplicationURL"          : notification_url,
            "ApplicationEnable"       : "Enable",
            "NotificationPayloadType" : "eBLSchemaSOAP",
            "DeviceType"              : "Platform",
        },
        UserDeliveryPreferenceArray={
            "NotificationEnable": notification_enables
        },
    )

    if response.Ack not in ("Success", "Warning"):
        for err in (response.Errors or []):
            raise RuntimeError(f"[{err.ErrorCode}] {err.LongMessage}")
    print(f"通知URL登録完了: {notification_url}")

def check_notification_preferences(config: dict, is_sandbox: bool = False) -> None:
    """GetNotificationPreferences で現在の通知設定を確認する"""
    session  = _make_session(config, "GetNotificationPreferences", is_sandbox)
    settings = Settings(strict=False, xml_huge_tree=True)
    client   = Client(wsdl=WSDL_URL, settings=settings,
                      transport=Transport(session=session))

    response = client.service.GetNotificationPreferences(
        RequesterCredentials={"eBayAuthToken": config["user_token"]},
        PreferenceLevel="UserData",
    )
    adp = response.ApplicationDeliveryPreferences
    print(f"  登録URL  : {adp.ApplicationURL}")
    print(f"  有効状態 : {adp.ApplicationEnable}")
    udpa = response.UserDeliveryPreferenceArray
    if udpa and udpa.NotificationEnable:
        for ne in udpa.NotificationEnable:
            print(f"  イベント : {ne.EventType} -> {ne.EventEnable}")

# ===== 実行例 =====
if __name__ == "__main__":
    config = {
        "app_id"     : "YourApp-XXXX",
        "dev_id"     : "your-dev-id-xxxx",
        "cert_id"    : "your-cert-id-xxxx",
        "user_token" : "AgAAAA**...",
    }
    EVENTS = [
        "AuctionCheckoutComplete",
        "FixedPriceTransaction",
        "FeedbackLeft",
        "ItemSold",
    ]
    register_notification_url(
        config=config,
        notification_url="https://your-server.example.com/ebay/notifications",
        events=EVENTS,
        is_sandbox=True,          # まずは Sandbox でテスト
    )
    check_notification_preferences(config, is_sandbox=True)

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

Platform Notifications の設定は一見シンプルですが、実務に投入すると必ず以下の壁にぶつかります。特に「なぜ通知が届かないのか」というデバッグは複数の原因が絡み合うため、数時間を費やす罠になりがちです。

1. 通知URLのHTTPS必須とチャレンジ・レスポンス検証

最も重要な制約は、通知 URL は本番環境では必ず HTTPS でなければならない点です（自己署名証明書は受け付けません。Let's Encrypt などの正規証明書が必要です）。Sandbox では http://localhost が一時的に許容される場合もありますが、本番では HTTPS のみです。

さらに、URL を登録しただけでは通知は届きません。eBay はまず GET リクエストを送信し、エンドポイントの正当性を確認する「チャレンジ・レスポンス検証」を行います。eBay は challenge_code というクエリパラメータ付きの GET を送り、サーバーは SHA-256(challenge_code + verificationToken + endpointURL) を計算した16進ハッシュを JSON で返さなければなりません。このレスポンスが正しくないと、SetNotificationPreferences の呼び出し自体は成功しても実際の通知は一切届きません。

さらに本番では、eBay が実際に通知を POST する際に X-EBAY-SIGNATURE ヘッダーが付与されます。このシグネチャを検証しないシステムは、悪意のある第三者がエンドポイントを叩いて注文処理を誤作動させるリスクを抱えます。本番システムでは署名検証を必ず実装してください。

2. 重複配信への対応（べき等性キーの設計）

eBay の通知は「At-Least-Once（少なくとも1回）配信」です。ネットワーク障害やサーバーの応答遅延が発生した場合、同一イベントが2回・3回と配信されることがあります。この前提を無視して「通知が来たら即座に注文レコードを作成する」実装をすると、同じ注文がデータベースに複数回書き込まれるという深刻なバグが発生します。

対策はべき等性（Idempotency）の確保です。eBay が送信する SOAP メッセージには Timestamp や ItemID / TransactionID が含まれており、これらを組み合わせた一意キーで「処理済みかどうか」を管理します。Redis の SET NX（Not eXists）コマンドを使い、処理開始時にキーを立て、成功後にそのキーを維持するパターンが堅牢です。

ポイントは、べき等性チェックを「処理完了後に記録する」のではなく「処理開始時にアトミックに取得する」ことです。処理中にクラッシュした場合に通知IDが「処理済み」として残ると、再送時にスキップされて注文が永遠に処理されない「処理漏れ」が発生します。失敗時はキーを削除してリトライを許可する設計が必要です。

3. SandboxとProductionで異なる環境設定の管理

eBay Sandbox と本番環境では API 呼び出し先が異なります（api.sandbox.ebay.com vs api.ebay.com）。通知も同様で、Sandbox 用の通知 URL と Production 用の通知 URL を環境変数で明確に分離して管理することが重要です。

よくある失敗が、「Sandbox 用に登録した URL に本番通知が届いてしまう（またはその逆）」という混線です。URL のパスに /sandbox/ や /production/ を含める命名規則を採用し、環境を判別しやすくするのが実務的なベストプラクティスです。設定をコードにハードコードせず、環境変数（.env や AWS Secrets Manager）から読み込む設計にしてください。

頻出エラーコード早見表

エラーコード	概要	対処法
21916667	DeliveryURL が無効な形式	URL の形式を確認。http:// は本番では不可。正規ドメインの https:// を使用する。
21916669	ApplicationURL must be https	http:// を https:// に変更。Let's Encrypt 等の正規 SSL 証明書が必要。
21916672	指定の EventType がサポート外	GetNotificationPreferences（PreferenceLevel=Application）で利用可能なイベント一覧を確認し、正確な文字列を指定する。
21916680	ApplicationURL の最大登録数を超過	GetNotificationPreferences で現在の登録数を確認し、不要なURLを削除してから再登録する。

堅牢な実装：FastAPIによるイベント振り分けエンドポイント

ここでは実務に耐える FastAPI ベースの通知受信サーバーを実装します。チャレンジ・レスポンス検証、イベント種別ごとの処理振り分け、エラーハンドリング、BackgroundTasks を使った即時応答を含む完全な実装です。事前に pip install fastapi uvicorn を実行してください。起動コマンドは uvicorn notification_receiver:app --host 0.0.0.0 --port 8000 です。

デコレータ方式の @notification_handler レジストリを採用することで、新しいイベント種別への対応を関数1つ追加するだけで拡張でき、コアのディスパッチロジックに変更を加えずに済みます。オープン・クローズド原則に従った設計です。

# notification_receiver.py
import hashlib
import logging
import os
import xml.etree.ElementTree as ET
from typing import Optional

from fastapi import FastAPI, Request, HTTPException, Query, BackgroundTasks
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(name)s %(message)s")

# 環境変数から設定を読み込む（本番は Secrets Manager 等を使用）
VERIFICATION_TOKEN: str = os.environ["EBAY_VERIFICATION_TOKEN"]   # 32文字以上推奨
ENDPOINT_URL: str       = os.environ["EBAY_NOTIFICATION_ENDPOINT_URL"]

app = FastAPI(title="eBay Platform Notification Receiver")

# ── イベントハンドラーレジストリ ──────────────────────────────
_HANDLERS: dict = {}

def notification_handler(event_type: str):
    """@notification_handler("AuctionCheckoutComplete") デコレータ"""
    def decorator(fn):
        _HANDLERS[event_type] = fn
        return fn
    return decorator

# ── GET /ebay/notifications  ← チャレンジ・レスポンス検証 ──────
@app.get("/ebay/notifications")
async def handle_challenge(challenge_code: Optional[str] = Query(default=None)):
    """
    eBay エンドポイント検証（チャレンジ・レスポンス）に応答する。

    SetNotificationPreferences で URL を登録すると、eBay はまず GET
    リクエストを送信してエンドポイントの正当性を確認する。
    challengeResponse = SHA-256(challenge_code + verificationToken + endpointUrl)
    """
    if challenge_code is None:
        raise HTTPException(status_code=400, detail="challenge_code が必要です")

    hash_input     = f"{challenge_code}{VERIFICATION_TOKEN}{ENDPOINT_URL}"
    challenge_resp = hashlib.sha256(hash_input.encode("utf-8")).hexdigest()
    logger.info(f"チャレンジ検証 code={challenge_code[:8]}... resp={challenge_resp[:8]}...")
    return JSONResponse(content={"challengeResponse": challenge_resp})

# ── POST /ebay/notifications ← 実際の通知受信 ─────────────────
@app.post("/ebay/notifications")
async def receive_notification(request: Request, bg: BackgroundTasks):
    """
    eBay からのプラットフォーム通知を受信し、イベント種別で振り分ける。

    重要: eBay は 30 秒以内に 200 応答を受け取れない場合に再送を行う。
    重い処理はすべて BackgroundTasks で非同期実行し、即座に 200 を返す。
    """
    body = await request.body()
    bg.add_task(_dispatch, body)
    return JSONResponse(content={"status": "accepted"}, status_code=200)

# ── 内部: SOAP XML 解析 → イベント振り分け ────────────────────
SOAP_NS = "http://schemas.xmlsoap.org/soap/envelope/"

async def _dispatch(body: bytes) -> None:
    """SOAPボディを解析し、登録済みハンドラーへ振り分ける"""
    try:
        root      = ET.fromstring(body)
        soap_body = root.find(f"{{{SOAP_NS}}}Body")
        if soap_body is None:
            logger.error("SOAPボディの解析失敗")
            return
        for child in soap_body:
            local_tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag
            handler   = _HANDLERS.get(local_tag)
            if handler:
                logger.info(f"通知受信: {local_tag}")
                await handler(child)
            else:
                logger.warning(f"未定義の通知タイプ: {local_tag}")
    except ET.ParseError as exc:
        logger.error(f"XML解析エラー: {exc} | body先頭={body[:120]}")

def _text(elem, path: str, default: str = "") -> str:
    """XML要素から安全にテキストを取得するヘルパー"""
    node = elem.find(path)
    return node.text if node is not None and node.text else default

# ── イベントハンドラー定義 ────────────────────────────────────
@notification_handler("AuctionCheckoutComplete")
async def handle_auction_checkout(elem) -> None:
    """オークション落札確定（AuctionCheckoutComplete）を処理する"""
    item_id  = _text(elem, "Item/ItemID")
    buyer_id = _text(elem, "Transaction/Buyer/UserID")
    txn_id   = _text(elem, "Transaction/TransactionID")
    logger.info(f"[AuctionCheckout] item={item_id} txn={txn_id} buyer={buyer_id}")
    # TODO: 注文DBへの保存、梱包・発送フロー起動

@notification_handler("FixedPriceTransaction")
async def handle_fixed_price_txn(elem) -> None:
    """固定価格取引完了（FixedPriceTransaction）を処理する"""
    item_id = _text(elem, "Item/ItemID")
    txn_id  = _text(elem, "Transaction/TransactionID")
    amount  = _text(elem, "Transaction/TransactionPrice", "0.00")
    logger.info(f"[FixedPriceTxn] item={item_id} txn={txn_id} amount=JPY{amount}")
    # TODO: 在庫数の減算、注文確認メール送信

@notification_handler("FeedbackLeft")
async def handle_feedback_left(elem) -> None:
    """フィードバック受信（FeedbackLeft）を処理する"""
    comment_type = _text(elem, "FeedbackDetail/CommentType")
    commenter    = _text(elem, "FeedbackDetail/CommentingUser")
    logger.info(f"[Feedback] type={comment_type} from={commenter}")
    # TODO: Negative/Neutral の場合は緊急アラートを発報

@notification_handler("ItemSold")
async def handle_item_sold(elem) -> None:
    """商品売却（ItemSold）を処理する"""
    item_id = _text(elem, "ItemID")
    logger.info(f"[ItemSold] item={item_id}")
    # TODO: 出品リストの更新、補充アラート

_HANDLERS レジストリとデコレータパターンにより、新しいイベント種別への対応は @notification_handler("新イベント名") を付けた async 関数を追加するだけです。コアの _dispatch ロジックに変更を加える必要がないため、機能拡張時の影響範囲を最小化できます。

receive_notification エンドポイントが即座に 200 を返し、重い処理を BackgroundTasks に委ねる設計は非常に重要です。eBay は 30 秒以内に 200 応答を受信できない場合に同一通知を再送します。重いハンドラーが同期で実行されると再送ループに入り、「重複通知の嵐」を引き起こします。さらに、ハンドラー内で例外が発生しても 500 を eBay に返してはいけません。500 は再送トリガーになるからです。すべての例外を内部で catch し、eBay には常に 200 を返す設計にしてください。

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

非同期キューとべき等性キー設計によるスケールアウト

FastAPI の BackgroundTasks はプロセス内での非同期処理であり、サーバーが1台の間は十分に機能します。しかし、月間1万件を超える注文を処理する規模になると単一プロセスでの処理はボトルネックになります。また、サーバーが突然クラッシュした場合、BackgroundTasks に積まれた未処理の通知が失われるリスクもあります。

本番規模のシステムでは、受信（FastAPI）と処理（ワーカー）を分離し、Celery + Redis（または Amazon SQS）を使ったメッセージキューアーキテクチャへの移行を強く推奨します。FastAPI はリクエストを受け取ったらキューに積んで即座に 200 を返し、複数の Celery ワーカーが並列でキューを消化するパターンです。これによりワーカーを水平スケールアウトして処理能力を動的に調整できます。

# tasks.py  ─  Celery + Redis によるべき等処理ワーカー
import logging
from celery import Celery
import redis

logger       = logging.getLogger(__name__)
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
celery_app   = Celery("ebay_notify",
                      broker="redis://localhost:6379/0",
                      backend="redis://localhost:6379/1")

IDEMPOTENCY_TTL = 86_400  # 24時間（秒）

@celery_app.task(bind=True, max_retries=3, default_retry_delay=10)
def process_fixed_price_transaction(self, notification_id: str, payload: dict) -> None:
    """
    固定価格取引通知をべき等性を保証しながら非同期処理する。

    Args:
        notification_id : 通知の一意ID（Timestamp + ItemID + TransactionID のハッシュ等）
        payload         : 通知 XML から抽出した取引データ
    """
    # ── Step1: べき等性チェック（SET NX = Not eXists） ──────────
    idem_key = f"ebay:processed:{notification_id}"
    acquired = redis_client.set(idem_key, "1", ex=IDEMPOTENCY_TTL, nx=True)
    if not acquired:
        logger.info(f"重複通知をスキップ: {notification_id}")
        return    # 既に処理済み → 安全に終了

    try:
        # ── Step2: ビジネスロジック ──────────────────────────────
        decrement_inventory(payload["item_id"], qty=1)
        order_id = create_order(payload)
        send_confirmation_email(payload["buyer_email"], order_id)
        logger.info(f"処理完了: notification={notification_id} order={order_id}")

    except Exception as exc:
        # 処理失敗時はべき等性キーを削除してリトライを許可
        redis_client.delete(idem_key)
        logger.error(f"処理失敗（リトライ予定）: {exc}")
        raise self.retry(exc=exc)

# FastAPI 側からはこう呼び出す:
# process_fixed_price_transaction.delay(notification_id, payload)

SET NX（Not eXists）は Redis のアトミック操作であるため、複数ワーカーが同じ notification_id を同時に処理しようとしてもどちらか一方だけが処理を進められることを保証します。分散システムにおける「競合状態（Race Condition）」の回避に不可欠なパターンです。

notification_id の生成方法も重要です。eBay の SOAP メッセージには固定の通知 ID フィールドが常に存在するわけではありません。Timestamp + ItemID + TransactionID を結合した文字列の SHA-256 ハッシュをキーとして使う設計が、重複排除において最も堅牢です。TTL を 24 時間に設定することで、Redis のメモリ消費を抑えつつ実運用上の再送ウィンドウをカバーできます。

まとめ

本記事では、定期ポーリングからイベント駆動アーキテクチャへの移行を実現する eBay Platform Notifications の全体像を解説しました。

• ベースライン: zeep を使った SetNotificationPreferences / GetNotificationPreferences の呼び出しにより、通知 URL とイベント種別（AuctionCheckoutComplete / FixedPriceTransaction / FeedbackLeft / ItemSold）を登録・確認する基本実装。
• 深いポイント: チャレンジ・レスポンス検証（SHA-256）によるエンドポイント認証、At-Least-Once 配信に対応したべき等性設計、Sandbox / Production の環境分離の重要性、および頻出エラーコードへの対処法。
• スケーリング: FastAPI の BackgroundTasks による即時応答パターンから、Celery + Redis を使ったキューベースのスケールアウト設計、および Redis SET NX を用いた分散環境でのべき等性保証。

Platform Notifications を導入することで、API コール消費をゼロに抑えながら注文確定から数秒以内に在庫更新・出荷指示・確認メールを起動できるリアルタイムシステムが実現します。

次のステップ

通知システムが整ったことで、eBay セラー業務における「リアクティブな処理」の基盤が完成しました。次は「プロアクティブな出品戦略」を支える基礎知識に目を向けます。

次回（#16）は、「GetCategoriesとGetCategoryFeaturesで出品カテゴリの構造とルールを取得する」です。eBay の膨大なカテゴリツリーをプログラムから走査し、対象カテゴリで必須となる Item Specifics や出品ルールを API で動的に取得する方法を解説します。お楽しみに！