Claude Code×Shopify・Stripe API【2026】EC運営と売上分析をAIエージェントで効率化

pip install requests stripe python-dotenv

# .env の中身
SHOPIFY_SHOP=your-dev-store
SHOPIFY_ADMIN_TOKEN=shpat_xxxxxxxx
SHOPIFY_API_VERSION=2026-07
STRIPE_API_KEY=rk_test_xxxxxxxx

ec-ops/
├── CLAUDE.md               # ルール・数字の定義
├── .env                    # トークン（Git・Claudeの対象外）
├── .claude/settings.json   # permissions / hooks
├── .claude/hooks/block_write_api.py
├── .claude/skills/monthly-ec-report/SKILL.md
├── scripts/shopify/        # client.py, fetch_products.py, check_products.py, fetch_orders.py
├── scripts/stripe/         # common.py, monthly_summary.py, mrr.py, failed_payments.py, reconcile.py
├── reports/                # 取得・集計結果＝事実（Git管理外）
├── proposals/              # 修正案CSV＝提案（人が承認）
└── drafts/                 # 説明文・フォロー文面の下書き

"""Shopify Admin GraphQL API の薄いクライアント（読み取り専用）"""
import os
import time

import requests
from dotenv import load_dotenv

load_dotenv()
VERSION = os.environ.get("SHOPIFY_API_VERSION", "2026-07")
ENDPOINT = f"https://{os.environ['SHOPIFY_SHOP']}.myshopify.com/admin/api/{VERSION}/graphql.json"
HEADERS = {"X-Shopify-Access-Token": os.environ["SHOPIFY_ADMIN_TOKEN"]}


def graphql(query: str, variables: dict | None = None) -> dict:
    if query.lstrip().lower().startswith("mutation"):
        raise ValueError("mutation is not allowed in this project")  # 書き込みはコードでも止める
    for attempt in range(5):
        resp = requests.post(ENDPOINT, json={"query": query, "variables": variables or {}},
                             headers=HEADERS, timeout=30)
        if resp.status_code == 429 or resp.status_code >= 500:
            time.sleep(2 ** attempt)
            continue
        resp.raise_for_status()
        body = resp.json()
        errors = body.get("errors") or []
        if any(e.get("extensions", {}).get("code") == "THROTTLED" for e in errors):
            time.sleep(2 ** attempt)  # クエリコストの上限に達したら待って再試行
            continue
        if errors:
            raise RuntimeError(errors)
        return body["data"]
    raise RuntimeError("retries exhausted")


def paginate(query: str, root: str, variables: dict | None = None):
    """pageInfo.endCursor を $cursor に渡して全ページをたどる"""
    variables = {**(variables or {}), "cursor": None}
    while True:
        conn = graphql(query, variables)[root]
        yield from conn["nodes"]
        if not conn["pageInfo"]["hasNextPage"]:
            return
        variables["cursor"] = conn["pageInfo"]["endCursor"]

"""商品とバリアントを reports/products.jsonl に保存する"""
import json

from client import paginate

QUERY = """
query Products($cursor: String) {
  products(first: 25, after: $cursor, query: "status:active OR status:draft") {
    pageInfo { hasNextPage endCursor }
    nodes {
      handle title status productType descriptionHtml tracksInventory totalInventory
      variants(first: 30) { nodes { sku price compareAtPrice } }
    }
  }
}
"""

with open("reports/products.jsonl", "w", encoding="utf-8") as f:
    for product in paginate(QUERY, "products"):
        f.write(json.dumps(product, ensure_ascii=False) + "\n")

"""ルールで商品を点検し、修正案CSVの雛形を作る（ストアには反映しない）"""
import csv
import json
import re
from collections import defaultdict
from decimal import Decimal
from statistics import median

MIN_DESC = 60  # 説明文が短すぎる目安（文字）
RATIO = 5      # 同じ商品タイプの中央値から5倍以上離れた価格は要確認

products = [json.loads(line) for line in open("reports/products.jsonl", encoding="utf-8")]
by_type = defaultdict(list)
for p in products:
    by_type[p["productType"]] += [Decimal(v["price"]) for v in p["variants"]["nodes"]]

issues = []  # (handle, sku, issue, detail)
for p in products:
    text = re.sub(r"<[^>]+>|\s", "", p["descriptionHtml"] or "")
    if len(text) < MIN_DESC:
        issues.append((p["handle"], "", "description_short", f"{len(text)}文字"))
    if p["status"] == "ACTIVE" and p["tracksInventory"] and (p["totalInventory"] or 0) <= 0:
        issues.append((p["handle"], "", "active_out_of_stock", "販売中で在庫0"))
    prices = by_type[p["productType"]]
    med = median(prices) if len(prices) >= 5 else None
    for v in p["variants"]["nodes"]:
        price = Decimal(v["price"])
        if price <= 0:
            issues.append((p["handle"], v["sku"], "price_zero", str(price)))
        if v["compareAtPrice"] and Decimal(v["compareAtPrice"]) <= price:
            issues.append((p["handle"], v["sku"], "compare_at_not_higher", f"{price} / {v['compareAtPrice']}"))
        if med and (price > med * RATIO or price * RATIO < med):
            issues.append((p["handle"], v["sku"], "price_outlier", f"price={price} median={med}"))

# proposed_value・reason は Claude Code が、approved_by は人が埋める
with open("proposals/fix_proposals.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.writer(f)
    w.writerow(["handle", "sku", "issue", "current", "proposed_value", "reason", "approved_by"])
    w.writerows([*i, "", "", ""] for i in issues)
print(f"{len(products)} products, {len(issues)} issues -> proposals/fix_proposals.csv")

"""Stripe共通：制限付きキーの確認、最小通貨単位の変換、JSTの月範囲"""
import os
from datetime import datetime, timedelta, timezone
from decimal import Decimal

import stripe
from dotenv import load_dotenv

load_dotenv()
stripe.api_key = os.environ["STRIPE_API_KEY"]
if not stripe.api_key.startswith("rk_"):
    raise SystemExit("読み取り専用の制限付きキー（rk_...）を設定してください")

JST = timezone(timedelta(hours=9))
# 小数のない通貨。JPY の 1000 は「1,000円」、USD の 1000 は「10.00ドル」
ZERO_DECIMAL = {"bif", "clp", "djf", "gnf", "jpy", "kmf", "krw", "mga",
                "pyg", "rwf", "ugx", "vnd", "vuv", "xaf", "xof", "xpf"}


def to_major(amount: int, currency: str) -> Decimal:
    return Decimal(amount) if currency.lower() in ZERO_DECIMAL else Decimal(amount) / 100


def month_range(ym: str) -> tuple[int, int]:
    y, m = map(int, ym.split("-"))
    start = datetime(y, m, 1, tzinfo=JST)
    end = datetime(y + m // 12, m % 12 + 1, 1, tzinfo=JST)
    return int(start.timestamp()), int(end.timestamp())

"""月次の売上・返金・新規/解約サブスクを通貨別に集計する"""
import csv
import sys
from collections import Counter

import stripe
from common import month_range, to_major

ym = sys.argv[1]  # 例: 2026-08
start, end = month_range(ym)
period = {"gte": start, "lt": end}
sales, refunds, new_subs, churned = Counter(), Counter(), Counter(), Counter()

for pi in stripe.PaymentIntent.list(created=period, limit=100).auto_paging_iter():
    if pi.status == "succeeded":
        sales[pi.currency] += pi.amount_received
for rf in stripe.Refund.list(created=period, limit=100).auto_paging_iter():
    if rf.status == "succeeded":
        refunds[rf.currency] += rf.amount
for sub in stripe.Subscription.list(status="all", created=period, limit=100).auto_paging_iter():
    if sub.status not in ("incomplete", "incomplete_expired"):
        new_subs[sub.currency] += 1
for sub in stripe.Subscription.list(status="canceled", limit=100).auto_paging_iter():
    if sub.ended_at and start <= sub.ended_at < end:  # canceled_at ではなく実際の終了日時
        churned[sub.currency] += 1

with open(f"reports/stripe_summary_{ym}.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.writer(f)
    w.writerow(["currency", "gross", "refunds", "net", "new_subs", "churned_subs"])
    for cur in sorted(set(sales) | set(refunds) | set(new_subs) | set(churned)):
        gross, ref = to_major(sales[cur], cur), to_major(refunds[cur], cur)
        w.writerow([cur.upper(), gross, ref, gross - ref, new_subs[cur], churned[cur]])

"""現在のMRRを通貨別に計算し、reports/mrr_history.csv に1行ずつ追記する（毎月末に実行）"""
import csv
from collections import Counter
from datetime import datetime
from decimal import Decimal

import stripe
from common import JST, to_major

TO_MONTHLY = {"day": Decimal(365) / 12, "week": Decimal(52) / 12, "month": Decimal(1), "year": Decimal(1) / 12}
mrr = Counter()

for sub in stripe.Subscription.list(status="active", limit=100).auto_paging_iter():
    for item in sub["items"]["data"]:  # sub.items は dict の items() と衝突するので添字で取る
        price = item["price"]
        rec = price.get("recurring")
        if not rec or price.get("unit_amount") is None:
            continue  # 従量・段階制などは対象外（定義はCLAUDE.md）
        amount = Decimal(price["unit_amount"]) * (item.get("quantity") or 1)
        mrr[sub.currency] += amount * TO_MONTHLY[rec["interval"]] / rec["interval_count"]

today = datetime.now(JST).date().isoformat()
with open("reports/mrr_history.csv", "a", newline="", encoding="utf-8") as f:
    for cur, value in sorted(mrr.items()):
        csv.writer(f).writerow([today, cur.upper(), round(to_major(int(value), cur), 2)])

"""直近30日の決済失敗を一覧化する。メール・氏名は出力せず顧客IDだけ残す"""
import csv
import time

import stripe
from common import to_major

since = {"gte": int(time.time()) - 30 * 86400}
rows = []
# 単発決済：失敗すると requires_payment_method に戻り、last_payment_error が残る
for pi in stripe.PaymentIntent.list(created=since, limit=100).auto_paging_iter():
    err = pi.last_payment_error
    if pi.status == "requires_payment_method" and err:
        rows.append([pi.id, pi.customer, to_major(pi.amount, pi.currency), pi.currency, err.get("decline_code")])
# サブスク：支払いに失敗した請求書は open のまま attempt_count が増える
for inv in stripe.Invoice.list(status="open", created=since, limit=100).auto_paging_iter():
    if inv.attempt_count:
        rows.append([inv.id, inv.customer, to_major(inv.amount_due, inv.currency), inv.currency, "invoice_failed"])

with open("reports/failed_payments.csv", "w", newline="", encoding="utf-8") as f:
    csv.writer(f).writerows([["id", "customer_id", "amount", "currency", "reason"], *rows])

query Orders($cursor: String, $q: String) {
  orders(first: 100, after: $cursor, query: $q, sortKey: CREATED_AT) {
    pageInfo { hasNextPage endCursor }
    nodes {
      id name createdAt displayFinancialStatus
      totalPriceSet { shopMoney { amount currencyCode } }
    }
  }
}
# 変数の例: {"q": "created_at:>=2026-08-01 created_at:<2026-09-01"}

"""Shopify注文とStripe入金を突合する。確定できないものは candidate / unmatched として人に回す"""
import csv
import sys
from datetime import datetime
from decimal import Decimal

import stripe
from common import month_range, to_major

KEY = "shopify_order_id"  # 連携側が metadata に入れているキー名（環境ごとに確認）
ym = sys.argv[1]
start, end = month_range(ym)
payments = [p for p in stripe.PaymentIntent.list(created={"gte": start, "lt": end}, limit=100).auto_paging_iter()
            if p.status == "succeeded"]
by_meta = {p.metadata.get(KEY): p for p in payments if p.metadata.get(KEY)}
used, out = set(), []

for o in csv.DictReader(open(f"reports/orders_{ym}.csv", encoding="utf-8")):
    amount = Decimal(o["total"])
    p = by_meta.get(o["order_id"].rsplit("/", 1)[-1])  # gid://shopify/Order/123 -> 123
    if p:
        ok = to_major(p.amount_received, p.currency) == amount
        out.append([o["name"], p.id, "matched" if ok else "amount_mismatch"])
        used.add(p.id)
        continue
    t = datetime.fromisoformat(o["created_at"].replace("Z", "+00:00")).timestamp()
    near = [p for p in payments if p.id not in used and abs(p.created - t) < 900
            and to_major(p.amount_received, p.currency) == amount]
    status = "candidate" if len(near) == 1 else ("ambiguous" if near else "unmatched_shopify")
    out.append([o["name"], near[0].id if len(near) == 1 else "", status])
    if len(near) == 1:
        used.add(near[0].id)
out += [["", p.id, "unmatched_stripe"] for p in payments if p.id not in used]

with open(f"reports/reconcile_{ym}.csv", "w", newline="", encoding="utf-8") as f:
    csv.writer(f).writerows([["shopify_order", "payment_intent", "status"], *out])

# CLAUDE.md（EC運営・売上分析）

## 目的
- Shopify・Stripeのデータを読み取り、点検・集計・下書きを作る。ストアや決済は変更しない

## データの取り方（APIを直接叩かず、必ずスクリプト経由）
- 商品点検: scripts/shopify/fetch_products.py → check_products.py
- 月次: scripts/stripe/monthly_summary.py YYYY-MM / MRR: mrr.py / 決済失敗: failed_payments.py
- 突合: scripts/shopify/fetch_orders.py YYYY-MM → scripts/stripe/reconcile.py YYYY-MM
- 金額の合計・比較は reports/ のCSVの値を使う。暗算で再集計しない

## 数字の定義
- 売上 = 成功した PaymentIntent の amount_received。返金はその月に作成された成功分
- MRR = 有効なサブスクの月額換算（年額は1/12）。割引・従量課金は含まない
- 解約 = その月に ended_at を迎えたサブスク。月の区切りは JST
- 金額は最小通貨単位（JPYは小数なし、USDはセント）→ common.to_major で変換済み

## 下書きのルール
- 商品説明: drafts/descriptions/<handle>.md。商品データにない効能・数値・比較を書かない
- 決済失敗のフォロー: drafts/followups/。顧客名は {{customer_name}} のまま。カード番号をメールで求めない
- 修正案: proposals/fix_proposals.csv の proposed_value と reason を埋める。approved_by は空のまま

## 禁止
- 返金・キャンセル・価格/在庫/顧客情報の変更を行うコードを書く・実行する
- .env を読む、reports/ の内容を外部に送る

{
  "permissions": {
    "allow": [
      "Bash(python scripts/shopify/fetch_products.py:*)",
      "Bash(python scripts/shopify/check_products.py:*)",
      "Bash(python scripts/shopify/fetch_orders.py:*)",
      "Bash(python scripts/stripe/monthly_summary.py:*)",
      "Bash(python scripts/stripe/mrr.py:*)",
      "Bash(python scripts/stripe/failed_payments.py:*)",
      "Bash(python scripts/stripe/reconcile.py:*)",
      "Read(./reports/**)",
      "Edit(./drafts/**)",
      "Edit(./proposals/**)"
    ],
    "ask": ["Edit(./scripts/**)"],
    "deny": [
      "Read(./.env)",
      "Bash(curl:*)",
      "Bash(wget:*)",
      "Bash(stripe:*)",
      "Bash(shopify:*)"
    ]
  },
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [{ "type": "command", "command": "python .claude/hooks/block_write_api.py" }]
      }
    ]
  }
}

"""PreToolUse フック：書き込み系APIを呼ぶコードがファイルに入るのを止める"""
import json
import re
import sys

PATTERNS = [r"stripe\.\w+\.(create|modify|delete|cancel|capture)\b", r"^\s*mutation\b"]

tool_input = json.load(sys.stdin).get("tool_input", {})
text = tool_input.get("content") or tool_input.get("new_string") or ""
for pattern in PATTERNS:
    if re.search(pattern, text, flags=re.MULTILINE):
        print("書き込み系APIはこのリポジトリでは実装しません（CLAUDE.md参照）", file=sys.stderr)
        sys.exit(2)  # exit 2 = ブロック。stderr が理由として Claude に返る

> 商品データを点検して、修正案CSVを作って。
  説明文が短い商品は上位5件だけ下書きもお願い。

● Bash(python scripts/shopify/fetch_products.py)
● Bash(python scripts/shopify/check_products.py)
  ⎿ 412 products, 37 issues -> proposals/fix_proposals.csv
● Read(proposals/fix_proposals.csv)

  37件の内訳（CSVの集計どおり）:
  description_short 19件 … 同じ仕入れ先の新商品に集中しています
  compare_at_not_higher 6件 … セール終了後に「通常価格」が残っている形です
  price_outlier 3件 … うち1件は 12000 が 120 になった桁ミスの疑い

● Edit(proposals/fix_proposals.csv)   ← proposed_value と reason を記入
● Write(drafts/descriptions/linen-shirt-navy.md) ほか4件

  ストアへの反映は行っていません。price_outlier は価格設定の意図を
  知っている方の確認が必要です。

> 8月の月次レポートと、決済失敗のフォロー文面の下書きを作って。

● Bash(python scripts/stripe/monthly_summary.py 2026-08)
● Bash(python scripts/stripe/failed_payments.py)
● Read(reports/stripe_summary_2026-08.csv) / Read(reports/mrr_history.csv) ...
● Write(reports/monthly_2026-08.md)
● Write(drafts/followups/expired_card.md)
● Write(drafts/followups/insufficient_funds.md)

  決済失敗は expired_card と insufficient_funds が中心です。
  Stripe側の自動再請求・通知メールと重複しないか確認してください。

# Shopify の開発者向けMCP（ドキュメント・GraphQLスキーマの参照。執筆時点）
claude mcp add shopify-dev -- npx -y @shopify/dev-mcp@latest
# Stripe のリモートMCP（URL・認証方式は公式ドキュメントで確認）
claude mcp add --transport http stripe https://mcp.stripe.com

---
name: monthly-ec-report
description: StripeとShopifyのデータから月次のEC売上レポート（売上・返金・解約・MRR・決済失敗・突合）を作る。月次レポートや売上集計を頼まれたときに使う。
---
1. 対象月 YYYY-MM を確認（指定がなければ先月）
2. monthly_summary.py → failed_payments.py → fetch_orders.py → reconcile.py の順に実行
3. reports/monthly_YYYY-MM.md に「サマリー3行／売上表／MRR推移／決済失敗の内訳／突合の不一致一覧」を書く
4. 数値はCSVから転記する。「事実」と「所見」を分け、原因は断定せず確認点として書く

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")/.."
YM=$(date -d "last month" +%Y-%m)  # GNU date の書き方
claude -p "monthly-ec-report の手順で ${YM} のレポートを作成して" \
  --allowedTools "Bash(python scripts/stripe/monthly_summary.py:*)" "Bash(python scripts/stripe/mrr.py:*)" \
                 "Bash(python scripts/stripe/failed_payments.py:*)" "Bash(python scripts/stripe/reconcile.py:*)" \
                 "Bash(python scripts/shopify/fetch_orders.py:*)" "Read" "Write(./reports/**)" \
  --output-format json > "logs/monthly_${YM}.json"