「AntigravityでAIエージェントを作ったが、自分しか使っていない」という状況は、実はもったいないことが多いです。
あなたが作った文書要約エージェント・コード生成エージェント・データ分析エージェントは、同じ問題を持つ企業や開発者にとって価値があるかもしれません。それをAPIとして公開し、使った分だけ課金する仕組みを作ることで、エージェント自体が収益を生み始めます。
ここではAntigravityエージェントをAPIサービスとして出荷するための設計と実装を、動くコードとともに解説します。
API公開の全体設計
エージェントをAPIとして公開する場合、最低限必要なコンポーネントは4つです。
① APIゲートウェイ: 認証・レート制限・リクエストルーティングを担当
② APIキー管理: ユーザーにAPIキーを発行し、ハッシュ化して保存する
③ 使用量計測: エージェントの呼び出し回数・トークン数を記録する
④ 課金統合: 使用量をStripeに報告し月次請求を自動化する
これらを一気に作ろうとすると大変ですが、順番に積み上げれば各コンポーネントは独立してテスト可能です。
Step 1: APIゲートウェイの実装
FastAPIとMiddlewareを使ったAPIゲートウェイの実装です。
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse
import hashlib
import time
import asyncio
from collections import defaultdict
from datetime import datetime, UTC
import anthropic
import uuid
app = FastAPI(title="AI Agent API")
claude_client = anthropic.Anthropic()
# インメモリのレート制限カウンター（本番ではRedisに移行）
rate_limit_store: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT_PER_MINUTE = 20 # プランごとに変える
class APIKeyAuth:
"""
APIキー認証の依存関係クラス。
本番ではDBからAPIキーを検索する。
"""
def __init__(self, db_lookup_fn=None):
# テスト用のインメモリストア
self._keys: dict[str, dict] = {
# key_hash -> {user_id, plan, rate_limit}
}
self.db_lookup = db_lookup_fn
def register_key(self, api_key: str, user_id: str, plan: str) -> str:
"""APIキーを登録し、ハッシュを保存"""
key_hash = hashlib.sha256(api_key.encode()).hexdigest()
self._keys[key_hash] = {
"user_id": user_id,
"plan": plan,
"rate_limit": self._plan_rate_limit(plan),
}
return key_hash
def _plan_rate_limit(self, plan: str) -> int:
return {"free": 5, "starter": 20, "pro": 100, "enterprise": 1000}.get(plan, 5)
async def __call__(self, request: Request) -> dict:
api_key = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").replace("Bearer ", "")
if not api_key:
raise HTTPException(status_code=401, detail="API key required")
key_hash = hashlib.sha256(api_key.encode()).hexdigest()
# DBルックアップ（テスト用インメモリにフォールバック）
user_info = self._keys.get(key_hash)
if not user_info:
raise HTTPException(status_code=401, detail="Invalid API key")
# レート制限チェック
user_id = user_info["user_id"]
now = time.time()
window = rate_limit_store[user_id]
# 60秒以上前のエントリを削除
rate_limit_store[user_id] = [t for t in window if now - t < 60]
if len(rate_limit_store[user_id]) >= user_info["rate_limit"]:
raise HTTPException(
status_code=429,
detail="Rate limit exceeded",
headers={"Retry-After": "60", "X-RateLimit-Limit": str(user_info["rate_limit"])}
)
rate_limit_store[user_id].append(now)
return user_info
auth = APIKeyAuth()
@app.get("/health")
async def health():
return {"status": "ok", "timestamp": datetime.now(UTC).isoformat()}
@app.post("/v1/agents/analyze")
async def analyze_document(
request: Request,
user_info: dict = Depends(auth),
):
"""
文書分析エージェントAPI。
リクエスト: {"text": str, "task": str}
レスポンス: {"result": str, "tokens_used": int, "request_id": str}
"""
body = await request.json()
text = body.get("text", "")
task = body.get("task", "要約")
if not text:
raise HTTPException(status_code=400, detail="'text' field is required")
if len(text) > 100_000:
raise HTTPException(status_code=400, detail="Text too long (max 100,000 characters)")
request_id = str(uuid.uuid4())
# Antigravityエージェントのロジック
response = claude_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=f"""あなたは文書分析の専門AIエージェントです。
与えられたテキストに対して、指定されたタスクを高精度で実行します。
JSON形式で結果を返してください。""",
messages=[
{
"role": "user",
"content": f"""タスク: {task}
テキスト:
{text}
結果を以下のJSON形式で返してください:
{{"result": "タスクの実行結果", "confidence": 0.0-1.0, "key_points": ["ポイント1", "ポイント2"]}}"""
}
]
)
tokens_used = response.usage.input_tokens + response.usage.output_tokens
# 使用量をDBに記録（非同期でバックグラウンド処理）
await record_usage(
user_id=user_info["user_id"],
request_id=request_id,
tokens=tokens_used,
endpoint="/v1/agents/analyze",
)
return {
"result": response.content[0].text,
"tokens_used": tokens_used,
"request_id": request_id,
}
async def record_usage(user_id: str, request_id: str, tokens: int, endpoint: str):
"""使用量を記録（実際はDB書き込み + Stripeへの報告キューイング）"""
print(f"Usage: user={user_id}, request={request_id}, tokens={tokens}, endpoint={endpoint}")
# DB書き込みとStripe報告のコードをここに追加Step 2: APIキーの発行と管理
セキュアなAPIキー発行システムです。
import secrets
import hashlib
import psycopg2
from datetime import datetime, UTC
def generate_api_key() -> tuple[str, str]:
"""
APIキーを生成する。
Returns: (raw_key, key_hash)
- raw_keyはユーザーに表示する（1回だけ）
- key_hashだけDBに保存する
"""
# 32バイトのランダムキー（base64 URL-safe）
raw_key = f"ak_{secrets.token_urlsafe(32)}"
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
return raw_key, key_hash
def create_api_key_for_user(
db_conn,
user_id: str,
key_name: str,
plan: str,
stripe_customer_id: str,
) -> str:
"""
ユーザーにAPIキーを発行してDBに保存。
raw_keyを返す（この後はハッシュしか保存しないため、1回だけ表示する）。
"""
raw_key, key_hash = generate_api_key()
with db_conn.cursor() as cur:
cur.execute("""
INSERT INTO api_keys (
id, user_id, key_hash, key_name, plan,
stripe_customer_id, is_active, created_at
) VALUES (
gen_random_uuid(), %s, %s, %s, %s,
%s, TRUE, NOW()
)
""", (user_id, key_hash, key_name, plan, stripe_customer_id))
db_conn.commit()
return raw_key # ユーザーへの表示はこの1回のみ
# スキーマ例
CREATE_TABLES_SQL = """
CREATE TABLE IF NOT EXISTS api_keys (
id UUID PRIMARY KEY,
user_id VARCHAR(255) NOT NULL,
key_hash VARCHAR(64) UNIQUE NOT NULL,
key_name VARCHAR(255),
plan VARCHAR(50) NOT NULL DEFAULT 'free',
stripe_customer_id VARCHAR(255),
stripe_subscription_item_id VARCHAR(255),
is_active BOOLEAN NOT NULL DEFAULT TRUE,
last_used_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS api_usage (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id VARCHAR(255) NOT NULL,
request_id VARCHAR(255) UNIQUE NOT NULL,
endpoint VARCHAR(255) NOT NULL,
tokens_used INTEGER NOT NULL DEFAULT 0,
billable_units INTEGER NOT NULL DEFAULT 0,
stripe_reported BOOLEAN NOT NULL DEFAULT FALSE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_api_usage_user ON api_usage(user_id, created_at);
CREATE INDEX idx_api_usage_unreported ON api_usage(stripe_reported) WHERE stripe_reported = FALSE;
"""Step 3: Stripe Metered Billingとの統合
使用量をStripeに報告するバックグラウンドジョブです。
import stripe
import psycopg2
stripe.api_key = "sk_live_..."
# トークン単位の課金設計
TOKENS_PER_BILLABLE_UNIT = 1000 # 1,000トークン = 1ユニット
PRICE_PER_UNIT_JPY = 2 # ¥2/1,000トークン（マークアップ込み）
def calculate_billable_units(tokens: int) -> int:
"""トークン数をStripeに報告するユニット数に変換"""
import math
return math.ceil(tokens / TOKENS_PER_BILLABLE_UNIT)
async def report_usage_batch(db_conn, stripe_subscription_item_id: str, user_id: str) -> int:
"""
DBの未報告使用量をStripeに送信する。
バックグラウンドジョブから5分ごとに呼び出すことを想定。
"""
with db_conn.cursor() as cur:
cur.execute("""
SELECT id, billable_units
FROM api_usage
WHERE user_id = %s
AND stripe_reported = FALSE
ORDER BY created_at ASC
LIMIT 200
FOR UPDATE SKIP LOCKED
""", (user_id,))
rows = cur.fetchall()
if not rows:
return 0
record_ids = [r[0] for r in rows]
total_units = sum(r[1] for r in rows)
try:
stripe.SubscriptionItem.create_usage_record(
stripe_subscription_item_id,
quantity=total_units,
timestamp=int(datetime.now(UTC).timestamp()),
action="increment",
)
cur.execute("""
UPDATE api_usage SET stripe_reported = TRUE
WHERE id = ANY(%s)
""", (record_ids,))
db_conn.commit()
return len(rows)
except stripe.error.StripeError as e:
db_conn.rollback()
print(f"Stripe report failed for user {user_id}: {e}")
raiseStep 4: プランとSLA設計
エンタープライズ顧客を獲得するためには、プランとSLAの設計が重要です。
from dataclasses import dataclass
@dataclass
class PlanConfig:
name: str
monthly_price_jpy: int
included_tokens: int # 基本料金に含まれるトークン数
rate_limit_per_minute: int
max_input_tokens: int # 1リクエストあたりの最大入力トークン
sla_uptime_percent: float # 稼働率SLA
support_level: str # none / email / priority
concurrent_requests: int # 同時リクエスト数
PLANS = {
"free": PlanConfig(
name="Free",
monthly_price_jpy=0,
included_tokens=10_000,
rate_limit_per_minute=5,
max_input_tokens=4_000,
sla_uptime_percent=99.0,
support_level="none",
concurrent_requests=1,
),
"starter": PlanConfig(
name="Starter",
monthly_price_jpy=2_980,
included_tokens=100_000,
rate_limit_per_minute=20,
max_input_tokens=20_000,
sla_uptime_percent=99.5,
support_level="email",
concurrent_requests=5,
),
"pro": PlanConfig(
name="Pro",
monthly_price_jpy=9_800,
included_tokens=500_000,
rate_limit_per_minute=100,
max_input_tokens=100_000,
sla_uptime_percent=99.9,
support_level="priority",
concurrent_requests=20,
),
"enterprise": PlanConfig(
name="Enterprise",
monthly_price_jpy=0, # 個別見積もり
included_tokens=0, # カスタム
rate_limit_per_minute=1000,
max_input_tokens=200_000,
sla_uptime_percent=99.95,
support_level="dedicated",
concurrent_requests=100,
),
}よくある落とし穴3つ
落とし穴1 — APIキーを平文で保存する
最も危険なミスです。APIキーはhashlib.sha256でハッシュ化し、ハッシュのみをDBに保存します。ユーザーへの表示は発行時の1回だけ。漏洩しても元のキーが復元できない設計にすることが必須です。
落とし穴2 — エラーレスポンスがAIの内部情報を漏らす
APIを公開する場合、内部エラーをそのまま返してはいけません。anthropic.APIErrorをキャッチして「Service temporarily unavailable」のような汎用メッセージに変換し、詳細は内部ログにのみ記録します。
落とし穴3 — SLAを設定したのに監視していない
99.9%稼働率を謳っているなら、実際にそれを計測する仕組みが必要です。UptimeRobot（無料プランあり）やBetterStackで1分ごとの死活監視を設定し、ダウン時に即通知が来るようにしてから公開することを強くお勧めします。
リリース前チェックリスト
エージェントAPIを公開する前に確認すべき項目を整理します。認証はAPIキーのハッシュ化が正しく機能しているか。レート制限は各プランで正しく動作しているか。使用量はDBに正しく記録されているか。Stripeへの報告ジョブは動作しているか。エラーハンドリングは内部情報を漏らさない設計になっているか。監視はダウン時に通知されるか。これらをすべて確認してから公開することで、初期の信頼を守れます。
最初の有料顧客は、初期設計よりも「しっかり動いている」という実績と応答速度が決め手になります。
個人開発12年とアーティスト活動から見るエージェント設計
線引きするときの3つの判断軸
- 失敗時の影響が金銭やユーザー体験にどれだけ波及するか
- 復旧オペレーションが明文化されているか
- 観測ログから人間が再現できる粒度に整っているか
まとめ — 検証ステップの推奨
- 観測メトリクスとアラートを設置
- 限定ロールアウトで本番検証