Claude APIを組み込んだサービスを作るとき、一番悩むのが課金モデルの設計ではないでしょうか。
月額固定プランはシンプルですが、ヘビーユーザーに吸われてコストが膨らむリスクがあります。従量制は公平ですが、ユーザーが使い方を読みにくく、チャーンにつながりやすい。その中間にある**使用量課金（Usage-Based Billing）**は、両者の良いところを取れる設計ですが、実装が複雑です。
ここではClaude APIのトークン計測から始まり、Stripe Metered Billingとの統合、ユーザーごとの利用制限まで、実際に動くコードと設計判断を丸ごと公開します。
使用量課金の設計思想：なぜフラットレートではないのか
使用量課金を選ぶ理由は、主に2つです。
コストとの整合性: Claude APIはトークン単位で課金されます。固定月額でサービスを提供すると、ヘビーユーザーがコストを食いつぶす構造になりがちです。使用量に比例した課金にすることで、マージンが安定します。
ユーザーにとっての公平感: 月に10回しか使わないユーザーと1,000回使うユーザーが同じ料金を払う構造は、前者が離脱する動機になります。使用量ベースにすることで、ライトユーザーを失わずに済みます。
課題は「何をどの単位で課金するか」の設計です。私がよく使うアプローチはクレジット制です。1クレジット = ¥1として内部管理し、Claude APIのコストに一定のマークアップを乗せてクレジットに換算します。ユーザーにはクレジットを購入してもらい、API呼び出しごとに消費させます。
Step 1: トークン使用量の計測とログ記録
Claude APIのレスポンスにはusageフィールドが含まれており、ここからinput/outputトークン数を取得できます。
import anthropic
import logging
from dataclasses import dataclass
from datetime import datetime, UTC
client = anthropic.Anthropic()
# ログ設定（本番環境ではCloud Logging等に出力）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class UsageRecord:
"""1回のAPI呼び出しの使用量記録"""
user_id: str
model: str
input_tokens: int
output_tokens: int
cache_creation_tokens: int # プロンプトキャッシュ作成
cache_read_tokens: int # プロンプトキャッシュ読み取り
timestamp: datetime
request_id: str
# モデルごとの価格（USD per 1M tokens）
MODEL_PRICING = {
"claude-opus-4-6": {
"input": 15.00,
"output": 75.00,
"cache_creation": 18.75, # input × 1.25
"cache_read": 1.50, # input × 0.10
},
"claude-sonnet-4-6": {
"input": 3.00,
"output": 15.00,
"cache_creation": 3.75,
"cache_read": 0.30,
},
"claude-haiku-4-5-20251001": {
"input": 0.80,
"output": 4.00,
"cache_creation": 1.00,
"cache_read": 0.08,
},
}
# JPY換算レート（本番では為替APIから取得推奨）
USD_TO_JPY = 148.0
# サービスのマークアップ倍率（2.5倍 = コストの2.5倍を顧客に請求）
MARKUP_FACTOR = 2.5
def calculate_cost_jpy(record: UsageRecord) -> float:
"""
UsageRecordからJPYコスト（顧客請求額）を計算する。
マークアップ込みの価格を返す。
"""
pricing = MODEL_PRICING.get(record.model)
if not pricing:
raise ValueError(f"Unknown model: {record.model}")
# USD単価（per 1M tokens）から実コストを計算
input_cost_usd = (record.input_tokens / 1_000_000) * pricing["input"]
output_cost_usd = (record.output_tokens / 1_000_000) * pricing["output"]
cache_create_usd = (record.cache_creation_tokens / 1_000_000) * pricing["cache_creation"]
cache_read_usd = (record.cache_read_tokens / 1_000_000) * pricing["cache_read"]
total_cost_usd = input_cost_usd + output_cost_usd + cache_create_usd + cache_read_usd
total_cost_jpy = total_cost_usd * USD_TO_JPY
# マークアップ適用（1円未満は切り上げ）
import math
billable_jpy = math.ceil(total_cost_jpy * MARKUP_FACTOR)
logger.info(
"usage_calculated",
extra={
"user_id": record.user_id,
"model": record.model,
"tokens_in": record.input_tokens,
"tokens_out": record.output_tokens,
"cost_usd": round(total_cost_usd, 6),
"billable_jpy": billable_jpy,
}
)
return billable_jpy
def call_claude_with_tracking(
user_id: str,
messages: list[dict],
model: str = "claude-sonnet-4-6",
system: str = "",
) -> tuple[str, UsageRecord]:
"""
Claude APIを呼び出し、レスポンスと使用量記録を返す。
呼び出し元はUsageRecordをDBに保存し、Stripeに報告する責任を持つ。
"""
import uuid
request_id = str(uuid.uuid4())
kwargs = {
"model": model,
"max_tokens": 4096,
"messages": messages,
}
if system:
kwargs["system"] = system
response = client.messages.create(**kwargs)
usage = response.usage
record = UsageRecord(
user_id=user_id,
model=model,
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
cache_creation_tokens=getattr(usage, 'cache_creation_input_tokens', 0),
cache_read_tokens=getattr(usage, 'cache_read_input_tokens', 0),
timestamp=datetime.now(UTC),
request_id=request_id,
)
return response.content[0].text, record
# 使用例
text, record = call_claude_with_tracking(
user_id="user_abc123",
messages=[{"role": "user", "content": "Pythonでシンプルなスタックを実装してください"}],
model="claude-sonnet-4-6",
)
cost_jpy = calculate_cost_jpy(record)
print(f"トークン: {record.input_tokens}入力 / {record.output_tokens}出力")
print(f"課金額（マークアップ込み）: ¥{cost_jpy}")このコードの設計で重要なのは、call_claude_with_trackingが使用量記録（UsageRecord）を返すだけで、DBへの保存やStripeへの報告は呼び出し元の責任にしていることです。関心の分離によって、テストが書きやすくなります。
Step 2: PostgreSQLでの使用量記録
使用量を正確に記録するためのDBスキーマです。
-- 使用量記録テーブル
CREATE TABLE api_usage_records (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id VARCHAR(255) NOT NULL,
request_id VARCHAR(255) UNIQUE NOT NULL,
model VARCHAR(100) NOT NULL,
input_tokens INTEGER NOT NULL DEFAULT 0,
output_tokens INTEGER NOT NULL DEFAULT 0,
cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
cost_jpy INTEGER NOT NULL DEFAULT 0, -- 円単位（整数）
stripe_reported BOOLEAN NOT NULL DEFAULT FALSE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_usage_user_id ON api_usage_records(user_id);
CREATE INDEX idx_usage_created_at ON api_usage_records(created_at);
CREATE INDEX idx_usage_stripe_reported ON api_usage_records(stripe_reported) WHERE stripe_reported = FALSE;
-- ユーザーの月次使用量集計ビュー
CREATE VIEW monthly_usage_by_user AS
SELECT
user_id,
DATE_TRUNC('month', created_at) AS month,
SUM(input_tokens) AS total_input_tokens,
SUM(output_tokens) AS total_output_tokens,
SUM(cost_jpy) AS total_cost_jpy,
COUNT(*) AS request_count
FROM api_usage_records
GROUP BY user_id, DATE_TRUNC('month', created_at);cost_jpyを整数（円単位）で保存しているのは、浮動小数点の計算誤差を避けるためです。金額を扱うテーブルでは、最小単位の整数で保存する方が安全です。
Step 3: Stripe Metered Billingとの統合
Stripeの使用量課金（Metered Billing）を使うと、月末に使用量を自動集計して請求できます。
import stripe
import psycopg2
from datetime import datetime, UTC
stripe.api_key = "sk_live_..." # 実際はシークレット管理ツールから取得
def report_usage_to_stripe(
stripe_subscription_item_id: str,
user_id: str,
db_conn
) -> int:
"""
未報告の使用量をStripeに送信する。
バックグラウンドジョブから定期的に呼び出すことを想定。
Returns: 報告した件数
"""
with db_conn.cursor() as cur:
# 未報告のレコードを取得（バッチサイズ上限100）
cur.execute("""
SELECT id, cost_jpy, created_at
FROM api_usage_records
WHERE user_id = %s
AND stripe_reported = FALSE
ORDER BY created_at ASC
LIMIT 100
FOR UPDATE SKIP LOCKED
""", (user_id,))
records = cur.fetchall()
if not records:
return 0
# 使用量の合計（Stripeには集計値で送信する方が効率的）
total_cost_jpy = sum(r[1] for r in records)
record_ids = [r[0] for r in records]
try:
# Stripeに使用量を報告
stripe.SubscriptionItem.create_usage_record(
stripe_subscription_item_id,
quantity=total_cost_jpy, # ¥1単位で送信
timestamp=int(datetime.now(UTC).timestamp()),
action="increment", # 加算モード
)
# DBを更新（stripe_reported = TRUE）
cur.execute("""
UPDATE api_usage_records
SET stripe_reported = TRUE
WHERE id = ANY(%s)
""", (record_ids,))
db_conn.commit()
return len(records)
except stripe.error.StripeError as e:
db_conn.rollback()
# エラーはログに記録し、次回のバッチで再試行
logger.error(f"Stripe usage report failed: {e}", extra={"user_id": user_id})
raise
def setup_stripe_metered_product() -> dict:
"""
Stripeで使用量課金商品をセットアップする（初回1回だけ実行）。
返り値にprice_idとproduct_idが含まれる。
"""
# 商品作成
product = stripe.Product.create(
name="AI APIサービス（使用量課金）",
description="Claude API使用量に応じた従量課金プラン",
)
# 価格（Metered Billing）作成
price = stripe.Price.create(
product=product.id,
currency="jpy",
billing_scheme="per_unit",
unit_amount=1, # ¥1/クレジット
recurring={
"interval": "month",
"usage_type": "metered", # 使用量課金
"aggregate_usage": "sum", # 月内の合計値で請求
},
)
return {"product_id": product.id, "price_id": price.id}action="increment"を使っているのは、複数のバッチ処理が並行して実行されたときに二重計上を防ぐためです。set（絶対値指定）よりincrement（加算）の方が冪等性を保ちやすい場面が多いです。
Step 4: ユーザーごとのプランと利用制限
使用量課金でも、プランによって上限を設けることはよくある設計です。
from enum import Enum
from typing import Optional
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
class PlanTier(str, Enum):
FREE = "free"
STARTER = "starter" # ¥980/月
PRO = "pro" # ¥3,800/月
ENTERPRISE = "enterprise" # 個別契約
# プランごとの月間クレジット上限
PLAN_LIMITS = {
PlanTier.FREE: 500, # ¥500相当
PlanTier.STARTER: 5_000, # ¥5,000相当（¥980の売上に対してコスト管理）
PlanTier.PRO: 30_000, # ¥30,000相当
PlanTier.ENTERPRISE: None, # 無制限（別途交渉）
}
# プランごとの使用可能モデル
PLAN_ALLOWED_MODELS = {
PlanTier.FREE: ["claude-haiku-4-5-20251001"],
PlanTier.STARTER: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6"],
PlanTier.PRO: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6", "claude-opus-4-6"],
PlanTier.ENTERPRISE: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6", "claude-opus-4-6"],
}
class UsageLimitExceeded(Exception):
"""使用量上限超過エラー"""
def __init__(self, limit: int, current: int, plan: PlanTier):
self.limit = limit
self.current = current
self.plan = plan
super().__init__(f"月間上限 ¥{limit} に達しました（現在 ¥{current}）")
class ModelNotAllowed(Exception):
"""プランで許可されていないモデルの使用"""
def __init__(self, model: str, plan: PlanTier):
self.model = model
self.plan = plan
super().__init__(f"モデル {model} は {plan.value} プランでは使用できません")
def get_monthly_usage_jpy(user_id: str) -> int:
"""Redisから今月の累計使用量を取得（高速キャッシュ）"""
now = datetime.now(UTC)
cache_key = f"usage:{user_id}:{now.year}:{now.month}"
cached = r.get(cache_key)
if cached is not None:
return int(cached)
# キャッシュミス時はDBから取得してキャッシュ
# （実際はDB接続を引数で受け取るか依存注入する）
usage = 0 # DB問い合わせ結果
r.setex(cache_key, 3600, usage) # 1時間キャッシュ
return usage
def increment_monthly_usage_cache(user_id: str, cost_jpy: int) -> int:
"""使用量をRedisに加算し、新しい合計を返す"""
now = datetime.now(UTC)
cache_key = f"usage:{user_id}:{now.year}:{now.month}"
new_total = r.incrby(cache_key, cost_jpy)
# TTLが設定されていなければ設定（月末まで）
if r.ttl(cache_key) == -1:
import calendar
days_in_month = calendar.monthrange(now.year, now.month)[1]
seconds_until_eom = (days_in_month - now.day + 1) * 86400
r.expire(cache_key, seconds_until_eom)
return new_total
def check_and_execute(
user_id: str,
plan: PlanTier,
model: str,
messages: list[dict],
system: str = "",
) -> tuple[str, int]:
"""
プランチェック → 上限確認 → API呼び出し → 使用量記録
を一連で実行する。APIレスポンスと課金額（JPY）を返す。
Raises:
ModelNotAllowed: プランで許可されていないモデルを指定した場合
UsageLimitExceeded: 月間上限を超えた場合
"""
# モデル許可チェック
if model not in PLAN_ALLOWED_MODELS[plan]:
raise ModelNotAllowed(model, plan)
# 上限チェック（上限ありのプランのみ）
plan_limit = PLAN_LIMITS[plan]
if plan_limit is not None:
current_usage = get_monthly_usage_jpy(user_id)
if current_usage >= plan_limit:
raise UsageLimitExceeded(plan_limit, current_usage, plan)
# API呼び出し
response_text, record = call_claude_with_tracking(
user_id=user_id,
messages=messages,
model=model,
system=system,
)
# 課金額計算
cost_jpy = calculate_cost_jpy(record)
# Redisキャッシュ更新（非同期でDBとStripeにも保存すること）
new_total = increment_monthly_usage_cache(user_id, cost_jpy)
logger.info(f"Usage: user={user_id}, cost=¥{cost_jpy}, monthly_total=¥{new_total}")
return response_text, cost_jpyよくある落とし穴3つ
実際に使用量課金を実装して気づいた問題を正直に書きます。
1. Stripeへの二重報告
バックグラウンドジョブが並行して動作したとき、同じレコードをStripeに2回報告してしまうことがあります。解決策はStep 2で示したようにFOR UPDATE SKIP LOCKEDを使い、取得中のレコードを他のジョブがスキップするようにすることです。
2. 為替レート固定の危険性
USD_TO_JPY = 148.0をハードコードしていますが、為替が10%変動するとマージンが大きくブレます。月次で為替APIから取得して更新する仕組みを入れることを強くお勧めします。exchangerate.hostやopenexchangerates.orgが使いやすいです。
3. Redisとの整合性
Redisのキャッシュが正確でない場合（Redis障害・再起動など）、実際の使用量より多くまたは少なく見えることがあります。毎日未明にRedisをDBから再構築するバッチジョブを入れておくと、長期的な整合性が保てます。
応用例：段階的プランのアップセル設計
上限に近づいたユーザーへのアップセルは、使用量課金の最大の強みです。
def get_upgrade_nudge(user_id: str, plan: PlanTier) -> Optional[str]:
"""
上限の80%を超えたら、アップグレード促進メッセージを返す。
APIレスポンスに付与して表示する。
"""
plan_limit = PLAN_LIMITS.get(plan)
if plan_limit is None:
return None # Enterpriseは上限なし
current = get_monthly_usage_jpy(user_id)
usage_ratio = current / plan_limit
if usage_ratio >= 0.95:
return f"今月の利用上限まで残り ¥{plan_limit - current}。Proプランへアップグレードすると上限が ¥{PLAN_LIMITS[PlanTier.PRO]:,} に拡大されます。"
elif usage_ratio >= 0.80:
return f"今月の利用量が上限の{int(usage_ratio * 100)}%に達しました。"
return Noneこの関数をAPIレスポンスのメタデータに含めておけば、フロントエンドでバナーを出すだけで自然なアップセルが実現します。
実装順序のまとめ
設計が複雑に見えますが、実装順序を整理すると次のようになります。まずStep 1とStep 2でトークン計測とDB記録を先に作り、実際のコストデータを蓄積します。次にStep 4の上限チェックを入れ、コストが無制限にならない安全弁を先に作ります。最後にStep 3のStripe Metered Billingを組み込み、実際の請求を自動化します。
最初から全部を完璧に作ろうとしないことが、リリースを早める秘訣です。計測だけ先にできていれば、課金の仕組みは後から追加できます。