第2回楽天商品をPythonのMCP Serverで取得・DB蓄積してみる

はじめに

第1回では、Codexと自作MCP Serverを組み合わせたAIアフィリエイト支援システムの全体像を設計しました。Codexが判断・文章生成を担当し、Affiliate MCP Serverが楽天API・PostgreSQL・WordPressなどの外部システムとの接続を担当する構成です。

第2回からは実装を開始します。今回は、楽天APIから商品情報を取得し、Pythonで作成したAffiliate MCP Serverを経由してPostgreSQLへ保存する「Product Tools」を実装します。ここで蓄積した商品データを、第3回の商品選定・キーワード選定・記事生成で利用します。

1. 今回のゴール

今回のゴールは、楽天APIから取得した商品情報をAffiliate MCP Server経由でPostgreSQLへ保存し、CodexからMCP Toolとして利用できる状態にすることです。

楽天API
   ↓
Affiliate MCP Server
   ↓
PostgreSQL

具体的には、次のToolを実装します。

Tool	役割
`search_products(keyword)`	キーワードで楽天商品検索APIを呼び出し、商品一覧を取得する
`get_ranking_products()`	楽天ランキングから売れ筋商品を取得する
`get_product(item_code)`	特定商品の基本情報を取得する
`save_product(product)`	取得した商品をPostgreSQLへ保存する（UPSERT）
`list_saved_products()`	保存済み商品を一覧取得する（確認用。第1回のTool設計に追加）

補足: 第1回の設計では get_product_detail() も挙げていました。これは楽天APIに同名のAPIが存在するという意味ではなく、商品説明・画像・詳細スペックなど、記事生成に必要な情報をまとめて取得するための拡張用MCP Toolとして想定したものです。

第2回では、商品名・価格だけでなく、商品画像URL・キャッチコピー・商品説明・レビュー情報など、記事生成で利用する基本情報も products テーブルへ保存します。また、商品カテゴリによって異なる詳細スペックを保存できるよう、product_attributes テーブルもあらかじめ用意します。

get_product_detail() は、第3回でより詳細な商品情報を記事生成に利用する場合に、products と product_attributes の情報をまとめて取得するToolとして利用できます。

開発環境（本シリーズの進め方）

本シリーズでは、開発はWindowsのローカル環境で完結します。

Windows 11
 │
 ├─ Codex CLI
 ├─ Python / venv
 ├─ Affiliate MCP Server
 │
 └─ Docker Desktop
      └─ PostgreSQL

開発・単体テストはWindows上で行います（コード作成、pip install、MCP Inspectorでの確認まで）。
MCP ServerはWindows上でPythonの仮想環境（venv）を使って直接実行します。
PostgreSQLのみ、Docker Desktop上のコンテナで動かします。
ソースコードはGitで管理します（バージョン管理のため）。認証情報を含む .env はGitに含めません（後述）。

この前提のもと、第2回では .gitignore・.gitattributes を用意し、.env に認証情報をまとめて置く構成にします。

文字化け・改行コード対策

将来の実行環境（Linuxなど）との差異を減らすため、最初にエンコーディングと改行コードを統一しておきます。これを怠ると、日本語コメントの文字化けや、\r（CR）混入によるシェルスクリプトの実行エラーが起きます。

ファイルはすべてUTF-8（BOMなし）で保存します。server.py / db.py / .env などのソースは、エディタでUTF-8を明示的に指定します（VS Codeなら右下のエンコード表示から UTF-8 を選択）。
改行コードはLFに統一します。リポジトリのルートに .gitattributes を置き、Gitで管理するテキストファイルの改行コードをLFに統一します。

# .gitattributes
* text=auto eol=lf
*.py text eol=lf
*.md text eol=lf
*.sh text eol=lf

.gitattributes もGitで管理します（後述の git add に含めます）。これにより、開発環境や将来の実行環境による改行コードの差異を減らせます。

本シリーズのTool実装ロードマップ

第1回で設計したToolを、どの回で実装するか整理します。データベースについては、第2回の段階で第3回までに必要となる3テーブル（products / product_attributes / articles）を作成します。本シリーズ内では、このDB構成を変更せずに使用します。

カテゴリ	Tool	実装する回	対象テーブル
Product	`search_products()`	第2回	products
Product	`get_ranking_products()`	第2回	products
Product	`get_product()`	第2回	products
Product	`save_product()`	第2回	products
Product	`list_saved_products()`	第2回	products
Article	`list_article_candidates()`	第3回	products / articles
Article	`check_duplicate_article()`	第3回	articles
Article	`save_article()`	第3回	articles
WordPress	`create_draft()`	第3回	articles
WordPress	`get_post()`	第3回	articles
WordPress	`update_draft()`	第3回	articles

2. プロジェクト構成

プロジェクト構成です。第2回と第3回でディレクトリ構造は変えません。第3回はファイルを移動・リネームせず、tools/ などにファイルを足すだけなので、ここでは第3回までを含めた形を先に示します。

affiliate-mcp-server/
│
├─ .venv/                 # Python仮想環境（Gitには含めない）
├─ .env                   # 認証情報（第3回でWordPress変数を追記）
├─ .gitignore
├─ .gitattributes
├─ server.py              # エントリポイント（MCP Server起動）
├─ mcp_app.py             # MCPServerインスタンスを保持する共有モジュール
├─ db.py                  # DB接続・テーブル作成
├─ rakuten.py             # 楽天APIの共通処理（呼び出し・変換）
├─ wordpress.py           # WordPress REST APIの共通処理（第3回で追加）
├─ tools/
│   ├─ __init__.py        # tools をパッケージにする（空でよい）
│   ├─ product_tools.py   # Product Tools（第2回）
│   ├─ article_tools.py   # Article Tools（第3回で追加）
│   └─ wordpress_tools.py # WordPress Tools（第3回で追加）
├─ requirements.txt
└─ compose.yaml           # 開発用PostgreSQL

第2回で作成するのは、このうち次のファイルです。

.env / .gitignore / .gitattributes / requirements.txt / compose.yaml
server.py / mcp_app.py / db.py / rakuten.py
tools/__init__.py / tools/product_tools.py

.venv/ は動作確認用のPython仮想環境です。インストールした依存パッケージが入るため、.env と同様にGitには含めません（.gitignore で除外します）。

mcp_app.py に MCPServer インスタンスを置き、server.py と各 tools/*.py の両方から参照します。こうすることで、Toolモジュールが server.py を import して発生する循環importを避けられます。

本シリーズではWindows上でMCP Serverを直接実行するため、Pythonの仮想環境（venv）を使用します。PostgreSQLのみDocker Desktop上で起動します。

GitHubからプロジェクトをクローンする

本記事で使用するプロジェクトは、GitHubで公開しています。
次のコマンドでリポジトリをクローンします。

git clone https://github.com/aiota-jp/affiliate-mcp-server.git

3. Python MCP Serverを準備する

必要なものは次の通りです。

Python（3.11以上を推奨）
MCP SDK（mcp）
httpx（楽天APIの呼び出し）
python-dotenv（.env の読み込み）
psycopg（PostgreSQL接続、psycopg3系）

requirements.txt は次のようにします。

mcp[cli]>=2,<3
httpx
python-dotenv
psycopg[binary]

開発環境（Windows）では、依存をシステムのPythonへ直接入れず、仮想環境（venv）を作って、その中で動作確認します。プロジェクト直下に .venv という名前で仮想環境を作成し、有効化します。

# Windows（PowerShell）
cd affiliate-mcp-server
python -m venv .venv
.\.venv\Scripts\Activate.ps1

補足: WindowsのPowerShellで Activate.ps1 の実行がポリシーで拒否される場合は、Set-ExecutionPolicy -Scope Process RemoteSigned を一度実行してから有効化します。

有効化するとプロンプトの先頭に (.venv) が付きます。この状態で依存をインストールします。

pip install -r requirements.txt

以降の python server.py やMCP Inspectorでの確認は、この (.venv) が有効な状態で行います。作業を終えるときは deactivate で仮想環境を抜けられます。

補足（venvを使う理由）: venvが必要なのは、システムのPythonを汚さないためです。本シリーズではWindows上でMCP Serverを直接実行するため、venvの中に依存パッケージを閉じ込めます。PostgreSQLはDockerコンテナで動かすので、venvの対象には含めません。

4. 環境変数を設定する

認証情報はコードに直書きせず、.env で管理してMCP Server側に閉じ込めます。

# .env
RAKUTEN_APP_ID=あなたのApplication ID
RAKUTEN_ACCESS_KEY=あなたのAccess Key
RAKUTEN_AFFILIATE_ID=あなたのAffiliate ID

# 楽天アフィリエイトに登録した運営サイトのURL（HTTPヘッダーとして送信）
RAKUTEN_REFERER=https://あなたの運営サイト/
RAKUTEN_ORIGIN=https://あなたの運営サイト

DATABASE_URL=postgresql://affiliate:password@localhost:5432/affiliate?client_encoding=UTF8

# PostgreSQL（compose.yaml から参照。DATABASE_URL と値を一致させる）
DATABASE_USER=affiliate
DATABASE_PASSWORD=password
DATABASE_NAME=affiliate

現行の楽天ウェブサービスAPIでは、applicationId（Application ID）と accessKey（Access Key）の両方が必須です。片方しか送らないとエラーになるため、.env には両方を設定します。

RAKUTEN_REFERER と RAKUTEN_ORIGIN には、楽天アフィリエイト側に登録した運営サイトのURLを設定します。今回の環境では、これらを付けずにAPIを呼び出すと 403 REQUEST_CONTEXT_BODY_HTTP_REFERRER_MISSING が返されたため、HTTPヘッダー（Referer / Origin）として送信します。

上記はWindowsの開発環境用の .env です。Codex / MCP ServerはWindowsホスト上で動き、PostgreSQLはDocker上で動くため、DATABASE_URL の接続先は localhost:5432 を指定します。

補足（DATABASE_URL のパスワード）: postgresql://affiliate:password@localhost:5432/affiliate の password の部分は、DBユーザー affiliate のパスワードです。5章で追加する DATABASE_PASSWORD（compose.yaml を通じてPostgreSQLコンテナの初期パスワードになる）と一致させる必要があります。ここに書いた password はサンプル用のプレースホルダです。次の点に注意してください。

本番では必ず推測されにくい強固なパスワードに変更します。とくに外部公開する場合は必須です（第1回の非機能要件Security）。開発用のローカル環境でも、そのまま使い回さないようにします。

パスワードに記号（@ : / # など）を含む場合は、URLエンコード（パーセントエンコード）が必要です。たとえば p@ss#1 は p%40ss%231 と書きます。エンコードしないと @ などがURLの区切り文字と解釈され、接続に失敗します。

パスワードを含む DATABASE_URL は認証情報なので、.env（.gitignore 済み）に置き、Gitへは絶対にコミットしません。

補足（日本語の文字化け対策）: 商品名は日本語を含むため、DBの文字コードはUTF-8にします。postgres の公式イメージはデフォルトでUTF-8ですが、接続側でも明示しておくと確実です。DATABASE_URL の末尾に ?client_encoding=UTF8 を付ける方法があります（例: postgresql://affiliate:password@localhost:5432/affiliate?client_encoding=UTF8）。

.env は必ず .gitignore に追加し、Gitへコミットしないようにします。

# .gitignore
.env
.venv/
__pycache__/
*.pyc

補足: Application ID（RAKUTEN_APP_ID）とAccess Key（RAKUTEN_ACCESS_KEY）は、楽天ウェブサービスでアプリを登録すると取得できます。アフィリエイトID（RAKUTEN_AFFILIATE_ID）は楽天アフィリエイトから取得できます。取得したリンクにアフィリエイトIDを含めることで、成果が計測されます。

5. PostgreSQLを準備する

DockerでPostgreSQLを起動する

開発環境（Windows）では、PostgreSQLをDocker Desktop上のコンテナで動かします。全体の呼び出しの流れは次の通りです。

Codex
  │ MCP（stdio）
  ▼
MCP Server（Python / MCPServer）
  │ search_products()
  ▼
psycopg（PostgreSQL用ドライバ）
  │ localhost:5432
  ▼
Docker
  └─ PostgreSQL
       └─ products テーブル

Codexはツールを呼ぶだけです。「その裏でDBに接続してSQLを実行している」ことは、MCP Server（Python）側が引き受けます。ここでも「文章や判断はCodex、正確なデータ取得はPython」という役割分担は変わりません。

まず、Docker Desktopを起動しておきます。

Windowsのスタートメニューから Docker Desktop を起動します。
タスクトレイのクジラのアイコンが「Docker Desktop is running」の状態になるまで待ちます。
PowerShellで、Dockerが使える状態かを確認します。

docker version

サーバー（Engine）側の情報まで表示されれば準備完了です（error during connect などが出る場合は、Docker Desktopがまだ起動しきっていません）。なお、postgres は公開イメージなので、ローカルで使うだけならDocker Desktopへのサインインは不要です。

compose.yaml を作成します。ここで注意したいのは、compose.yaml はGitに含めるという点です。パスワードを直書きすると、.env を .gitignore で守っていても、compose.yaml 経由で認証情報がリポジトリに上がってしまいます。そこで、DBのユーザー名・パスワード・DB名は compose.yaml に直書きせず、.env から ${...} で読み込みます（Docker Composeは同じディレクトリの .env を自動で読み込みます）。

# compose.yaml
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: ${DATABASE_USER}
      POSTGRES_PASSWORD: ${DATABASE_PASSWORD}
      POSTGRES_DB: ${DATABASE_NAME}
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

参照される値は .env に追加します。
DATABASE_URL のユーザー名・パスワード・DB名と一致させます。

# .env（4章の内容に追記）
DATABASE_USER=affiliate
DATABASE_PASSWORD=password
DATABASE_NAME=affiliate

補足: この .env は、アプリ用の変数（RAKUTEN_APP_ID / DATABASE_URL など）と、Docker Composeが参照する DATABASE_USER / DATABASE_PASSWORD / DATABASE_NAME を1ファイルに同居させる形になります。これは一般的な使い方です。DATABASE_URL（アプリがDBへ接続するURL）と DATABASE_USER / DATABASE_PASSWORD / DATABASE_NAME（コンテナ初期化用）は別々の変数なので、両方 .env に置き、値（affiliate / password / affiliate）を一致させます。

PowerShellで、PostgreSQLをバックグラウンド起動します。

docker compose up -d

これで localhost:5432 にPostgreSQLが立ち上がり、.env の DATABASE_URL=postgresql://affiliate:password@localhost:5432/affiliate?client_encoding=UTF8 でそのまま接続できます。volumes でデータを永続化しているため、docker compose down してもデータは残ります（完全に消す場合は docker compose down -v）。

補足: この compose.yaml は開発用のPostgreSQL（db）だけを定義しています。MCP ServerはWindows上でvenvを使って直接実行します。パスワードはサンプルなので、そのまま使い回さず、強固な値に変更してください（第1回の非機能要件Security）。

テーブルを作成する

本シリーズでは、第2回の時点で第3回までに必要となる products・product_attributes・articles の3テーブルを定義します。第2回では主に products を使用し、第3回では保存した商品情報をもとに記事を生成して articles へ保存します。本シリーズ内では、このDB構成を変更せずに使用します。

テーブル	第2回で作成	主に使う回	役割
products	○	第2回〜	楽天の商品基本情報・画像・説明
product_attributes	○	第3回（必要時）	商品ごとに異なる詳細スペック・属性
articles	○	第3回	AIが生成した記事

-- 商品の基本情報（第2回〜）
CREATE TABLE IF NOT EXISTS products (
    id              SERIAL PRIMARY KEY,
    item_code       TEXT UNIQUE NOT NULL,
    name            TEXT NOT NULL,
    price           INTEGER,
    url             TEXT,
    affiliate_url   TEXT,
    -- 記事生成で利用する情報
    image_urls      TEXT[],
    catchcopy       TEXT,
    description     TEXT,
    -- 商品分類・評価
    genre_id        TEXT,
    review_count    INTEGER DEFAULT 0,
    review_average  NUMERIC(3, 2) DEFAULT 0,
    created_at      TIMESTAMP DEFAULT now(),
    updated_at      TIMESTAMP DEFAULT now()
);

-- 商品ごとに異なる詳細スペック・属性（第3回、必要時）
CREATE TABLE IF NOT EXISTS product_attributes (
    id               SERIAL PRIMARY KEY,
    product_id       INTEGER NOT NULL REFERENCES products(id) ON DELETE CASCADE,
    attribute_name   TEXT NOT NULL,
    attribute_value  TEXT,
    created_at       TIMESTAMP DEFAULT now(),
    UNIQUE (product_id, attribute_name)
);

-- AIが生成した現在の記事（第3回〜）
CREATE TABLE IF NOT EXISTS articles (
    id                SERIAL PRIMARY KEY,
    product_id        INTEGER NOT NULL REFERENCES products(id),
    title             TEXT NOT NULL,
    keyword           TEXT,
    content           TEXT,
    wordpress_post_id INTEGER,
    status            TEXT DEFAULT 'draft',
    published_at      TIMESTAMP,
    created_at        TIMESTAMP DEFAULT now(),
    updated_at        TIMESTAMP DEFAULT now()
);

各テーブルの役割は次の通りです。

products：楽天の商品基本情報を保存するテーブルです。商品コード・商品名・価格・URLに加えて、記事生成で利用する image_urls（商品画像URL、複数可）・catchcopy（キャッチコピー）・description（商品説明文）・レビュー情報なども保持します。商品画像は必須とはせず、取得できた場合のみ image_urls に保存します。楽天市場商品検索APIでは、商品名、キャッチコピー、価格、商品説明、商品URL、アフィリエイトURLなどが正式な出力項目として定義されています。
product_attributes：商品カテゴリによって異なる詳細スペック（例: CPU = Core Ultra 7）を、属性名・属性値のペアで保存するテーブルです。UNIQUE (product_id, attribute_name) で同一属性の重複登録を防ぎます。第3回で、より詳細な商品情報を記事生成に利用する場合に使います。なお、現在の楽天市場商品検索APIには attributeFlag があり、ジャンルIDを指定した場合に属性情報を返す機能があります。また、属性名などを取得する「楽天市場属性検索API」も提供されているため、将来 product_attributes を使う拡張方針自体は自然です。
articles：AIが生成した記事。content（本文）や、公開ステータス status・WordPressの投稿ID wordpress_post_id を持ちます。第3回で使用します。

db.py に、接続と全テーブルの作成をまとめておきます。外部キーの関係があるため、products → product_attributes → articles の順で作成します。

# db.py
import os
import psycopg
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.environ["DATABASE_URL"]


def get_connection():
    return psycopg.connect(DATABASE_URL)


def init_db():
    """本シリーズで使用するテーブルを作成する。"""
    with get_connection() as conn, conn.cursor() as cur:
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS products (
                id              SERIAL PRIMARY KEY,
                item_code       TEXT UNIQUE NOT NULL,
                name            TEXT NOT NULL,
                price           INTEGER,
                url             TEXT,
                affiliate_url   TEXT,
                image_urls      TEXT[],
                catchcopy       TEXT,
                description     TEXT,
                genre_id        TEXT,
                review_count    INTEGER DEFAULT 0,
                review_average  NUMERIC(3, 2) DEFAULT 0,
                created_at      TIMESTAMP DEFAULT now(),
                updated_at      TIMESTAMP DEFAULT now()
            )
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS product_attributes (
                id               SERIAL PRIMARY KEY,
                product_id       INTEGER NOT NULL
                                 REFERENCES products(id) ON DELETE CASCADE,
                attribute_name   TEXT NOT NULL,
                attribute_value  TEXT,
                created_at       TIMESTAMP DEFAULT now(),
                UNIQUE (product_id, attribute_name)
            )
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS articles (
                id                SERIAL PRIMARY KEY,
                product_id        INTEGER NOT NULL REFERENCES products(id),
                title             TEXT NOT NULL,
                keyword           TEXT,
                content           TEXT,
                wordpress_post_id INTEGER,
                status            TEXT DEFAULT 'draft',
                published_at      TIMESTAMP,
                created_at        TIMESTAMP DEFAULT now(),
                updated_at        TIMESTAMP DEFAULT now()
            )
            """
        )
        conn.commit()

本シリーズでは、第2回の時点で第3回までに必要となる3テーブルを定義します。第2回では主に products を使用し、第3回では保存した商品情報をもとに記事を生成して articles へ保存します。本シリーズ内では、このDB構成を変更せずに使用します。

6. Product Toolsを実装する（取得・保存）

この章では、Product Tools（商品の検索・ランキング取得・1件取得・保存・一覧）をまとめて実装します。処理の流れは次の通りです。

Codex
 ↓ 検索キーワード / ランキング要求
search_products() / get_ranking_products() / get_product()
 ↓
商品一覧・商品情報
 ↓
save_product() → PostgreSQL（products）

共通モジュール（`mcp_app.py` / `rakuten.py`）

まず、Toolの器になる mcp_app.py（MCP Serverインスタンス）を用意します。

# mcp_app.py
from mcp.server.mcpserver import MCPServer

# 各 tools/*.py と server.py はこの mcp を共有する（循環import回避）。
mcp = MCPServer("affiliate-mcp-server")

次に、楽天API呼び出しの共通処理を rakuten.py にまとめます。楽天APIのレスポンス（itemCode / itemName など）を、DBの列名（item_code / name など）へ変換する処理もここに置きます。これが「MCP Serverは境界」という責務分離です。

# rakuten.py
import os
import sys
import httpx
from dotenv import load_dotenv

# .env を読み込む
load_dotenv()

# 楽天APIの認証・接続情報
RAKUTEN_APP_ID = os.environ.get("RAKUTEN_APP_ID", "")
RAKUTEN_ACCESS_KEY = os.environ.get("RAKUTEN_ACCESS_KEY", "")
RAKUTEN_AFFILIATE_ID = os.environ.get("RAKUTEN_AFFILIATE_ID", "")

# 楽天アフィリエイトに登録した運営サイトURL
RAKUTEN_REFERER = os.environ.get("RAKUTEN_REFERER", "")
RAKUTEN_ORIGIN = os.environ.get("RAKUTEN_ORIGIN", "")

# URL末尾の日付はAPIの「バージョン番号」。
# エンドポイント系列・バージョンはAPIごとに異なるので、各APIの公式ドキュメントに従う。
SEARCH_URL = (
    "https://openapi.rakuten.co.jp/ichibams/api/IchibaItem/Search/20260701"
)
RANKING_URL = (
    "https://openapi.rakuten.co.jp/ichibaranking/api/IchibaItem/Ranking/20220601"
)


def search(url: str, params: dict) -> dict:
    """楽天市場APIを呼び出し、JSONを辞書で返す共通関数。"""
    # 必須の認証・接続情報を確認する
    if not RAKUTEN_APP_ID:
        raise RuntimeError(
            "RAKUTEN_APP_ID が未設定です。.env を確認してください。"
        )
    if not RAKUTEN_ACCESS_KEY:
        raise RuntimeError(
            "RAKUTEN_ACCESS_KEY が未設定です。.env を確認してください。"
        )
    if not RAKUTEN_REFERER:
        raise RuntimeError(
            "RAKUTEN_REFERER が未設定です。.env を確認してください。"
        )
    if not RAKUTEN_ORIGIN:
        raise RuntimeError(
            "RAKUTEN_ORIGIN が未設定です。.env を確認してください。"
        )

    # 楽天APIへ送信するパラメータ
    query = {
        "applicationId": RAKUTEN_APP_ID,
        "accessKey": RAKUTEN_ACCESS_KEY,
        "format": "json",
        **params,
    }
    # Affiliate ID が設定されている場合だけ追加
    if RAKUTEN_AFFILIATE_ID:
        query["affiliateId"] = RAKUTEN_AFFILIATE_ID

    # 楽天アフィリエイトに登録した運営サイトURLをHTTPヘッダーとして送信する
    headers = {
        "Referer": RAKUTEN_REFERER,
        "Origin": RAKUTEN_ORIGIN,
    }

    try:
        with httpx.Client(timeout=10.0, headers=headers) as client:
            res = client.get(url, params=query)
            # HTTPエラー（4xx/5xx）の場合
            if res.is_error:
                error_message = (
                    f"楽天APIエラー: status={res.status_code}, "
                    f"response={res.text}"
                )
                print(error_message, file=sys.stderr)
                raise RuntimeError(error_message)
            return res.json()
    # タイムアウトの場合
    except httpx.TimeoutException as e:
        error_message = f"楽天APIへの接続がタイムアウトしました: {e}"
        print(error_message, file=sys.stderr)
        raise RuntimeError(error_message) from e
    # 通信エラーの場合
    except httpx.RequestError as e:
        error_message = f"楽天APIへの接続に失敗しました: {e}"
        print(error_message, file=sys.stderr)
        raise RuntimeError(error_message) from e


def to_product(item: dict) -> dict:
    """楽天APIのレスポンスをDBの列名へ変換する（APIの都合をここで吸収する）。"""
    # 商品画像URL（mediumImageUrls）。無い商品もあるので空リストで受ける。
    image_urls = [
        img.get("imageUrl", "")
        for img in item.get("mediumImageUrls", [])
        if img.get("imageUrl")
    ]
    return {
        "item_code": item["itemCode"],
        "name": item["itemName"],
        "price": item.get("itemPrice"),
        "url": item.get("itemUrl"),
        "affiliate_url": item.get("affiliateUrl") or item.get("itemUrl"),
        "image_urls": image_urls,
        "catchcopy": item.get("catchcopy", ""),
        "description": item.get("itemCaption", ""),  # 商品説明文
        "genre_id": str(item.get("genreId", "")),
        "review_count": item.get("reviewCount", 0),
        "review_average": item.get("reviewAverage", 0),
    }

補足（商品URLについて）: 楽天APIのレスポンスでは、アフィリエイトIDを指定して取得した場合、環境やAPIによって itemUrl にも hb.afl.rakuten.co.jp から始まるアフィリエイト形式のURLが返る場合があります（今回確認したレスポンスでは、itemUrl もアフィリエイト形式で、その中に pc= パラメータとして通常の商品URLが含まれていました）。本記事ではAPIから返された itemUrl を url、affiliateUrl を affiliate_url としてそのまま保存します（URLを独自に解析して pc= を取り出す処理は行いません）。第3回の記事生成では、成果計測のため affiliate_url を商品リンクとして使用します。

RAKUTEN_REFERER / RAKUTEN_ORIGIN を Referer / Origin ヘッダーとして送信しています。今回の環境ではこれらが無いと 403 REQUEST_CONTEXT_BODY_HTTP_REFERRER_MISSING が返りました。timeout=10.0 は、第1回の非機能要件（Reliability）で触れた「応答が返ってこない場合に待ち続けない」ためです。あわせて、HTTPエラー・タイムアウト・通信エラーを捕捉し、原因が分かるメッセージを標準エラー出力へ出してから RuntimeError として上位に伝えます（同じくReliability / Observability）。

エンドポイントとバージョン番号: URL末尾の日付（例: 20260701）は、その日の日付ではなく、APIのバージョン番号（2026年7月1日版）です。20260914 のように毎日変えるものではなく、楽天が新バージョンを公開し古いバージョンが廃止されるタイミングでのみ更新します。

ただし、バージョン番号もエンドポイント系列もAPIごとに異なります。実際、この記事の2つのAPIでも違います。

API エンドポイントバージョン

商品検索 openapi.rakuten.co.jp/ichibams/... 20260701（2026-07-01版）

ランキング openapi.rakuten.co.jp/ichibaranking/... 20220601（2022-06-01版）

上記コードは formatVersion を指定していないため、デフォルトの formatVersion=1 で動きます。この形式ではレスポンスが Items → Item の入れ子構造になるため、entry["Item"] で1件ずつ取り出しています。将来 formatVersion=2 を使う場合は、レスポンスが各要素に直接フィールドを持つ形になるため、entry["Item"] ではなく entry を渡すように変更が必要です。

Product Tools（`tools/product_tools.py`）

Product Toolsは1つのファイル tools/product_tools.py にまとめて実装します。mcp_app の mcp を import して @mcp.tool() で登録し、楽天APIは rakuten の共通処理、DBアクセスは db を使います。各Toolの役割は次の通りです。

search_products()：キーワードで商品を検索する（hits は楽天APIの範囲 1〜30 に丸める）
get_ranking_products()：売れ筋ランキングを取得する
get_product()：item_code で1件の基本情報を取得する
save_product()：商品を products テーブルへ保存する（UPSERT）
list_saved_products()：保存済み商品を一覧取得する（確認用）

# tools/product_tools.py
from mcp_app import mcp
import rakuten
import db


@mcp.tool()
def search_products(keyword: str, hits: int = 20) -> list[dict]:
    """キーワードで楽天商品検索APIを呼び出し、商品一覧を返す。"""
    if not keyword.strip():
        raise ValueError("検索キーワードを指定してください。")

    # 楽天商品検索APIの hits は 1〜30 の範囲。CodexやユーザーがhitsにAPI範囲外の
    # 値（hits=100 など）を渡しても壊れないように、ここで丸める。
    hits = max(1, min(hits, 30))

    data = rakuten.search(rakuten.SEARCH_URL, {"keyword": keyword, "hits": hits})
    return [rakuten.to_product(e["Item"]) for e in data.get("Items", [])]


@mcp.tool()
def get_ranking_products(genre_id: str | None = None) -> list[dict]:
    """楽天ランキングから売れ筋商品を取得する。genre_id で絞り込みも可能。"""
    params = {}
    if genre_id:
        params["genreId"] = genre_id

    data = rakuten.search(rakuten.RANKING_URL, params)
    return [rakuten.to_product(e["Item"]) for e in data.get("Items", [])]


@mcp.tool()
def get_product(item_code: str) -> dict | None:
    """item_code で楽天商品検索APIを呼び出し、該当商品の基本情報を返す。"""
    if not item_code.strip():
        raise ValueError("商品コード（itemCode）を指定してください。")

    data = rakuten.search(rakuten.SEARCH_URL, {"itemCode": item_code})
    items = data.get("Items", [])
    if not items:
        return None
    return rakuten.to_product(items[0]["Item"])


@mcp.tool()
def save_product(product: dict) -> dict:
    """商品を products テーブルへ保存する（UPSERT）。"""
    with db.get_connection() as conn, conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO products (
                item_code, name, price, url, affiliate_url,
                image_urls, catchcopy, description,
                genre_id, review_count, review_average
            )
            VALUES (
                %(item_code)s, %(name)s, %(price)s, %(url)s, %(affiliate_url)s,
                %(image_urls)s, %(catchcopy)s, %(description)s,
                %(genre_id)s, %(review_count)s, %(review_average)s
            )
            ON CONFLICT (item_code) DO UPDATE SET
                name           = EXCLUDED.name,
                price          = EXCLUDED.price,
                url            = EXCLUDED.url,
                affiliate_url  = EXCLUDED.affiliate_url,
                image_urls     = EXCLUDED.image_urls,
                catchcopy      = EXCLUDED.catchcopy,
                description    = EXCLUDED.description,
                genre_id       = EXCLUDED.genre_id,
                review_count   = EXCLUDED.review_count,
                review_average = EXCLUDED.review_average,
                updated_at     = now()
            RETURNING id, item_code
            """,
            product,
        )
        row = cur.fetchone()
        conn.commit()
    return {"id": row[0], "item_code": row[1]}


@mcp.tool()
def list_saved_products(limit: int = 20) -> list[dict]:
    """保存済みの商品を一覧取得する（確認用）。"""
    # limit=-1 など想定外の値が渡っても壊れないように、1〜100 に丸める。
    limit = max(1, min(limit, 100))
    with db.get_connection() as conn, conn.cursor() as cur:
        cur.execute(
            """
            SELECT id, item_code, name, price, review_count, review_average
            FROM products
            ORDER BY updated_at DESC
            LIMIT %s
            """,
            (limit,),
        )
        rows = cur.fetchall()
    return [
        {
            "id": r[0],
            "item_code": r[1],
            "name": r[2],
            "price": r[3],
            "review_count": r[4],
            "review_average": float(r[5]) if r[5] is not None else 0,
        }
        for r in rows
    ]

認証情報のチェックとHTTP呼び出しは rakuten.search() に、レスポンス変換は rakuten.to_product() にまとめてあるので、各Toolはパラメータの組み立てと結果の整形だけに集中できます。save_product() / list_saved_products() はDBを使うため、ファイル先頭で import db しています。

エントリポイント（`server.py`）

最後に、エントリポイントの server.py を用意します。mcp_app を読み込み、Toolモジュール（tools.product_tools）をimportして登録し、DBを初期化してからサーバーを起動します。Toolはimportされた時点で @mcp.tool() により登録されるため、tools.product_tools をimportすることが重要です（第3回では、ここに tools.article_tools などのimportを足していきます）。

# server.py
from mcp_app import mcp
import db

import tools.product_tools

if __name__ == "__main__":
    db.init_db()
    mcp.run()

補足: tools/ をパッケージとして確実にimportできるよう、空の tools/__init__.py を置いておくと安全です。

重複データを防止する（UPSERT）

同じ商品を何度取得しても、DBに重複行が増えないようにします。第1回の非機能要件（Reliability）の冪等性（Idempotency）を、DBレベルで担保する部分です。

同じ item_code
   ↓
既存なら UPDATE
新規なら INSERT

save_product() の ON CONFLICT (item_code) DO UPDATE がこれを実現しています。products.item_code の UNIQUE 制約と組み合わせることで、「ランキング取得を毎日実行しても、同じ商品は最新情報で更新されるだけ」になります。

7. MCP Inspectorで確認する

実装したToolは、Codexにつなぐ前に、MCP Inspectorで単体テストできます。

npx @modelcontextprotocol/inspector .\.venv\Scripts\python.exe server.py

Inspector上で、search_products に keyword=ノートPC を渡す、get_ranking_products を引数なしで実行する、といった単体確認を行い、期待通りの商品一覧が返るかを確認します。DB保存は save_product を実行後、list_saved_products で反映を確認します。

画面上部の「Tools」をクリックしてください。今回実装した、「search_products」「get_ranking_products」「get_product」「save_product」「list_saved_products」の5つが表示されます。
続けてget_ranking_productsを選択し「Execute Tool」ボタンを押します。

実行結果

8. Codexから実行する

MCP Inspectorで各Toolが正常に動作することを確認できたら、次は作成したMCP ServerをCodex CLIへ登録します。

MCP ServerをCodexへ登録する

MCP Inspectorを起動している場合は、いったん Ctrl + C で終了しておきます。次に、作成したMCP ServerをCodexへ登録します。

codex mcp add affiliate -- .\.venv\Scripts\python.exe .\server.py

登録を確認する

登録内容を確認します。

codex mcp list

affiliate が表示され enabled になっていれば登録済みです。続けてCodexを起動します。

codex

Codexの画面で /mcp を実行し、affiliate が接続済みで、今回作成したToolが認識されていることを確認します。登録される主なToolは次の5つで、いずれも tools/product_tools.py で @mcp.tool() を付けたものです。

Codexから楽天ランキングを取得する

接続できたら、Codexへ自然言語で依頼します。CodexがAPI失敗時に別経路（Web検索など）でランキングを取得してしまうことがあるため、動作確認では使うMCP Toolを明示します。

affiliate MCP Serverの get_ranking_products ツールを使って、
楽天ランキングの上位5商品を取得してください。
別の方法では取得しないでください。

Codexは内容を判断し、MCP Serverの get_ranking_products() を呼び出します。Tool実行時に許可を求める画面が表示された場合は、内容を確認して許可します。正常に実行されれば、楽天ランキングAPIから取得した商品情報がCodexへ返されます。

取得した商品をPostgreSQLへ保存する

成功を確認したら、取得から保存までまとめて依頼します。ここでもMCP Serverを明示します。

affiliate MCP Serverを使って、
楽天ランキングから上位5商品を取得し、
PostgreSQLへ保存してください。

実行結果

Codexはこの依頼をそのままSQLへ変換するわけではなく、次の流れで必要なMCP Toolを組み合わせて実行します。

ユーザー
  ↓「ランキング上位5商品を保存して」
Codex
  ↓
get_ranking_products()
  ↓
楽天API
  ↓ 商品データ
Codex
  ↓
save_product() × 5
  ↓
PostgreSQL（products テーブル）

get_ranking_products() は楽天ランキングAPIから商品を取得し、save_product() は受け取った商品情報をPostgreSQLへUPSERTします。

保存結果を確認する

最後にCodexへ、次のように依頼します。

affiliate MCP Serverを使って、
PostgreSQLに保存されている商品を一覧表示してください。

実行結果

Codexが list_saved_products() を呼び出し、products テーブルに保存されている商品を取得できれば確認完了です。これで、次の一連の処理がつながりました。

自然言語
   ↓
Codex
   ↓ MCP
Affiliate MCP Server
   ├─ get_ranking_products()
   ├─ save_product()
   └─ list_saved_products()
        ↓
     PostgreSQL

まとめ（第2回の完成状態）

今回作った商品収集・保存基盤は、次の流れがつながった状態です。

楽天API
   ↓
Affiliate MCP Server
   │
   └─ Product Tools
        ↓
PostgreSQL
   ├─ products
   ├─ product_attributes
   └─ articles

※ 第2回では主に products へ商品情報を保存

第2回では、楽天APIから商品を取得してPostgreSQLへ蓄積する、商品収集・保存基盤を実装しました。

項目	ポイント
何を作ったか	楽天API → Affiliate MCP Server → PostgreSQL の商品収集・保存基盤
DB	開発用の `compose.yaml`（`db` サービス）でPostgreSQLをDocker起動。本シリーズで使用する products / product_attributes / articles の3テーブルを第2回で作成。今回は主に products に商品情報を蓄積し、product_attributes / articles は第3回で利用する
実装したTool	`search_products()` / `get_ranking_products()` / `get_product()` / `save_product()` / `list_saved_products()`
ファイル構成	Toolをカテゴリ別に `tools/` へ分割（第2回は `product_tools.py`）。`mcp_app.py` でMCPServerインスタンスを共有し、循環importを回避。第3回はファイルを足すだけ
責務分離	楽天APIのフィールド名をDB列名へ変換するのは `rakuten.py`（境界）の役割
APIエンドポイント	商品検索は `ichibams/.../Search/20260701`、ランキングは `ichibaranking/.../Ranking/20220601`。URL末尾はバージョン番号（毎日変える値ではなく、APIごとに異なる）
認証情報	`applicationId` と `accessKey` の両方が必須。運営サイトURLを `Referer` / `Origin` ヘッダーとして送信（今回の環境では、未指定時に403エラーを確認）。`.env` に閉じ込め、`.gitignore` でGitに含めない。`compose.yaml` はパスワードを直書きせず `.env`（`DATABASE_USER` / `DATABASE_PASSWORD` / `DATABASE_NAME`）を `${...}` で参照（Gitに上げても安全）
重複防止	`item_code` の UNIQUE 制約 + UPSERT（ON CONFLICT）で冪等性を担保
単体確認	MCP Inspectorで各ToolをテストしてからCodexへつなぐ
開発環境	WindowsのvenvでMCP Serverを実行、PostgreSQLはDocker Desktopで起動。ソースはGit管理、`.venv` と `.env` はGitに含めない
文字化け対策	UTF-8で統一（ソース保存・DB `client_encoding=UTF8`）、改行は `.gitattributes` で LF に固定

次回・第3回では、ここで蓄積した商品をCodexに評価・選定させ、キーワードを決めて記事を生成し、WordPressへ下書き登録するところまでを実装します。第3回でArticle ToolsとWordPress Toolsを追加し、Affiliate MCP Serverを完成させます。

シリーズ記事一覧

LinuxにWordPressを構築して独自ドメインでHTTPS公開するまで ― インストール手順とシェルスクリプトによる自動化
 【実践第1回】Codex + MCPでAIアフィリエイト支援システムを設計してみる〜商品情報取得・DB蓄積・記事作成・WordPress連携〜
【実践第2回】楽天商品をPythonのMCP Serverで取得・DB蓄積してみる〜商品検索・ランキング取得・保存まで〜
【実践第3回】Codexに商品・キーワードを選定させてアフィリエイト記事を作成してみる〜商品評価・キーワード選定・記事生成・WordPress下書き登録まで〜

参考

※ 楽天APIの仕様・エンドポイント・規約は更新されることがあります。実装前に公式の最新情報を確認してください。掲載コードは学習用のサンプルです。実運用では、今回実装したタイムアウト・基本的なエラーハンドリングに加えて、リトライ、構造化ログ、監視などを追加すると、より安定した運用ができます（第1回の非機能要件）。