「AIエージェントでサービスを作りたいが、どう収益化するかイメージが湧かない」という声はよく聞きます。技術的には作れても、サブスクリプションの仕組みや課金設計、そして「継続して使ってもらえるエージェント」の設計が難しいのは事実です。
私は2025年後半にAgentKit（当時v1系）を使った業務自動化エージェントサービスを立ち上げ、半年でMRR180万円まで成長させました。その過程で学んだ「エージェントサービスの収益設計」と、最新のAgentKit 2.0を使った実装方法を、この記事で共有します。
AgentKit 2.0 の新機能と収益化への影響
AgentKit 2.0（2026年3月リリース）では、サービス化に直結する重要な機能がいくつか追加されました。
永続的なエージェントメモリ: セッションをまたいで文脈を保持できるようになりました。これにより「先週の続き」から自然に作業を再開できます。ユーザーが継続使用するインセンティブになり、解約率が下がります。
マルチエージェント協調の簡素化: 複数のエージェントを直感的にオーケストレーションできるAPIが整備されました。専門特化したサブエージェントを組み合わせることで、複雑な業務フローを処理できます。
コスト管理API: エージェントの実行コスト（トークン使用量・API呼び出し数）をリアルタイムで取得できます。テナント別のコスト管理が大幅に楽になりました。
# AgentKit 2.0 の基本構造
from agentkit import Agent, Memory, CostTracker
from agentkit.integrations import StripeWebhookHandler
# 永続メモリを持つエージェントの作成
class CustomerSuccessAgent(Agent):
"""顧客の業務を学習しながら継続的に改善するエージェント"""
def __init__(self, tenant_id: str, stripe_subscription_id: str):
super().__init__(
model="gemma4:27b", # ローカルモデルでAPIコスト削減
memory=Memory(
backend="postgresql",
tenant_id=tenant_id,
retention_days=365 # 1年間の文脈を保持
),
cost_tracker=CostTracker(
tenant_id=tenant_id,
billing_id=stripe_subscription_id
)
)
self.tenant_id = tenant_id
async def process_task(self, task: str, context: dict = None) -> dict:
"""
タスクを受け取り、過去の文脈を参照しながら処理する
Returns: {"result": str, "cost_tokens": int, "memory_updated": bool}
"""
# 過去の文脈を取得
relevant_memories = await self.memory.retrieve_relevant(task, top_k=5)
# エージェントの実行
result = await self.run(
task=task,
context={**(context or {}), "past_context": relevant_memories}
)
# 重要な学習を記憶に保存
await self.memory.store(
content=result.summary,
importance=result.importance_score,
tags=result.extracted_tags
)
return {
"result": result.output,
"cost_tokens": result.tokens_used,
"memory_updated": result.stored_new_memories
}サブスクリプション設計：3つの課金モデル
エージェントサービスの課金モデルは、ツール型SaaSとは異なる考え方が必要です。
モデル1: 実行回数ベース（シンプルで計算しやすい）
エージェントタスクの実行数に応じて課金します。顧客側のコスト予測が立てやすく、導入障壁が低いです。
Starterプラン ¥30,000/月 — 月500タスクまで
Growthプラン ¥80,000/月 — 月3,000タスクまで
Enterpriseプラン ¥200,000/月 — 月無制限
超過分 ¥50/タスク
適切なプラン決定のヒント:
- 小規模企業（〜20名）: Starterで十分なケースが多い
- 中規模企業（〜100名）: Growth推奨
- 大手・高頻度ユースケース: Enterprise
モデル2: 価値ベース（削減コスト連動型）
エージェントが削減した業務時間に基づいて課金するモデルです。ROIが明確で高い単価を正当化しやすいですが、計測が複雑です。
# 価値ベース課金の実装例
class ValueBasedBilling:
"""エージェントが削減した業務コストに基づく課金"""
HOURLY_LABOR_RATE = 2500 # 顧客の人件費（円/時間）の想定値
COMMISSION_RATE = 0.25 # 削減コストの25%を徴収
def calculate_monthly_fee(self, tenant_id: str, month: str) -> dict:
# エージェントが実行したタスクの記録を取得
tasks = db.get_tasks(tenant_id=tenant_id, month=month)
total_minutes_saved = sum(t.estimated_manual_minutes for t in tasks)
hours_saved = total_minutes_saved / 60
labor_cost_saved = hours_saved * self.HOURLY_LABOR_RATE
billing_amount = int(labor_cost_saved * self.COMMISSION_RATE)
# 最低課金額と最高課金額の設定
billing_amount = max(billing_amount, 50_000) # 最低¥50,000
billing_amount = min(billing_amount, 500_000) # 最高¥500,000
return {
"tenant_id": tenant_id,
"tasks_completed": len(tasks),
"hours_saved": round(hours_saved, 1),
"labor_cost_saved": labor_cost_saved,
"billing_amount": billing_amount,
"roi_multiplier": round(labor_cost_saved / billing_amount, 1)
}モデル3: 座席数ベース（チーム導入向け）
ユーザー数に応じた課金は、チーム全体への導入を進めたい場合に有効です。
Stripe との完全統合実装
サブスクリプションの管理にはStripeが最も信頼性が高く、AgentKit 2.0との統合も簡単です。
# stripe_integration.py
import stripe
from datetime import datetime
import logging
stripe.api_key = "sk_live_YOUR_KEY"
logger = logging.getLogger(__name__)
class AgentKitStripeIntegration:
"""AgentKit 2.0 × Stripe サブスクリプション管理"""
PRICE_IDS = {
"starter": "price_1ABC...starter",
"growth": "price_1ABC...growth",
"enterprise": "price_1ABC...enterprise"
}
OVERAGE_PRICE_ID = "price_1ABC...overage" # 使用量ベース
def create_subscription(self,
email: str,
plan: str,
tenant_id: str) -> dict:
"""新規顧客のサブスクリプションを作成する"""
try:
# Stripe Customer の作成
customer = stripe.Customer.create(
email=email,
metadata={"tenant_id": tenant_id, "plan": plan}
)
# サブスクリプションの作成（試用期間14日）
subscription = stripe.Subscription.create(
customer=customer.id,
items=[
{"price": self.PRICE_IDS[plan]},
# 従量課金項目（超過タスク用）
{
"price": self.OVERAGE_PRICE_ID,
"quantity": 0 # 初期値0、月次で更新
}
],
trial_period_days=14,
metadata={"tenant_id": tenant_id}
)
logger.info(f"Subscription created: {subscription.id} for {email}")
return {
"success": True,
"subscription_id": subscription.id,
"customer_id": customer.id,
"trial_end": datetime.fromtimestamp(subscription.trial_end).isoformat()
if subscription.trial_end else None
}
except stripe.error.StripeError as e:
logger.error(f"Stripe error for {email}: {e}")
return {"success": False, "error": str(e)}
def report_overage_usage(self, subscription_id: str,
overage_tasks: int) -> bool:
"""月次で超過タスク数をStripeに報告する"""
if overage_tasks <= 0:
return True
try:
# サブスクリプションの従量課金アイテムを特定
subscription = stripe.Subscription.retrieve(subscription_id)
overage_item = next(
(item for item in subscription.items.data
if item.price.id == self.OVERAGE_PRICE_ID),
None
)
if not overage_item:
logger.warning(f"No overage item found for {subscription_id}")
return False
# 使用量を記録
stripe.SubscriptionItem.create_usage_record(
overage_item.id,
quantity=overage_tasks,
timestamp=int(datetime.now().timestamp()),
action="set"
)
logger.info(f"Reported {overage_tasks} overage tasks for {subscription_id}")
return True
except stripe.error.StripeError as e:
logger.error(f"Failed to report overage: {e}")
return False
def handle_webhook(self, payload: bytes, sig_header: str) -> dict:
"""Stripeウェブフックを処理する"""
webhook_secret = "whsec_YOUR_SECRET"
try:
event = stripe.Webhook.construct_event(
payload, sig_header, webhook_secret
)
except stripe.error.SignatureVerificationError:
return {"status": "invalid_signature"}
handlers = {
"customer.subscription.created": self._handle_subscription_created,
"customer.subscription.deleted": self._handle_subscription_cancelled,
"customer.subscription.updated": self._handle_subscription_updated,
"invoice.payment_failed": self._handle_payment_failed,
"invoice.payment_succeeded": self._handle_payment_succeeded,
}
handler = handlers.get(event.type)
if handler:
handler(event.data.object)
return {"status": "ok", "event_type": event.type}
def _handle_subscription_created(self, subscription):
tenant_id = subscription.metadata.get("tenant_id")
# エージェントテナントを有効化
agent_registry.activate_tenant(tenant_id)
logger.info(f"Tenant activated: {tenant_id}")
def _handle_subscription_cancelled(self, subscription):
tenant_id = subscription.metadata.get("tenant_id")
# 解約理由の自動収集（チャーン分析用）
churn_analyzer.record_cancellation(tenant_id, subscription)
agent_registry.deactivate_tenant(tenant_id, grace_period_days=3)
logger.info(f"Tenant deactivating: {tenant_id}")
def _handle_payment_failed(self, invoice):
# 支払い失敗時のエスカレーション処理
customer_id = invoice.customer
send_payment_retry_email(customer_id, invoice.id)解約率を下げるエージェント設計のパターン
サブスクリプションビジネスで最も重要な指標は解約率（チャーン率）です。エージェントサービス特有の解約防止設計を3つ紹介します。
パターン1: 「いなくなると困る」ステータスを作る
エージェントが顧客の業務データを学習・蓄積することで、解約後に「データが消える損失」が意識されます。
class ChurnPreventionMemory:
"""解約防止のための価値蓄積型メモリシステム"""
async def generate_value_report(self, tenant_id: str) -> str:
"""エージェントが蓄積した価値を可視化するレポート"""
stats = await self.get_tenant_stats(tenant_id)
report = f"""
# エージェントが積み上げてきた成果（{stats['months_active']}ヶ月）
## 学習した業務パターン: {stats['learned_patterns']}件
## 処理した業務タスク: {stats['tasks_completed']:,}件
## 削減した作業時間: {stats['hours_saved']:.0f}時間
## 蓄積した業界知識: {stats['knowledge_entries']:,}エントリ
このデータはご解約時に失われます。
エクスポート機能は Enterpriseプランでご利用いただけます。
"""
return report.strip()パターン2: 月次「エージェントレポート」で価値を可視化する
async def send_monthly_value_report(tenant_id: str):
"""毎月1日に自動送付するROIレポート"""
stats = await get_monthly_stats(tenant_id)
# Claude APIでレポートを自然言語で生成
import anthropic
client = anthropic.Anthropic()
report_text = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=800,
messages=[{
"role": "user",
"content": f"""以下のデータをもとに、サービスの月次価値レポートを作成してください。
顧客目線で「このサービスを使い続けることの価値」が伝わるよう、
具体的な数字と感謝の気持ちを込めて書いてください。
先月の実績:
- 処理タスク数: {stats['tasks_completed']}件
- 削減した業務時間: {stats['hours_saved']:.0f}時間
- 主な自動化内容: {stats['top_automations']}
- 今月の新しい学習: {stats['new_patterns']}件"""
}]
).content[0].text
send_email(
to=get_tenant_email(tenant_id),
subject=f"【月次レポート】先月のエージェント実績",
body=report_text
)パターン3: 解約シグナルを早期検知して先手を打つ
class ChurnSignalDetector:
"""解約の前兆を検知してカスタマーサクセスが動けるようにする"""
CHURN_SIGNALS = [
{"metric": "weekly_tasks", "threshold_ratio": 0.5, "severity": "high"},
{"metric": "login_days", "threshold_ratio": 0.3, "severity": "high"},
{"metric": "error_rate", "threshold_value": 0.15, "severity": "medium"},
{"metric": "support_tickets", "threshold_value": 3, "severity": "medium"},
]
async def analyze_tenant_health(self, tenant_id: str) -> dict:
current = await get_current_metrics(tenant_id)
baseline = await get_baseline_metrics(tenant_id) # 過去3ヶ月平均
signals = []
for signal in self.CHURN_SIGNALS:
metric = signal["metric"]
if "threshold_ratio" in signal:
ratio = current[metric] / max(baseline[metric], 1)
if ratio < signal["threshold_ratio"]:
signals.append({
"type": metric,
"severity": signal["severity"],
"current": current[metric],
"baseline": baseline[metric],
"ratio": ratio
})
health_score = 100 - (len([s for s in signals if s["severity"] == "high"]) * 25 +
len([s for s in signals if s["severity"] == "medium"]) * 10)
if health_score < 60:
await alert_customer_success(tenant_id, signals, health_score)
return {"health_score": health_score, "signals": signals}よくある失敗パターン：エージェントSaaSを壊す3つのミス
ミス1: トークンコストを課金に反映させない
AgentKit 2.0でGemma 4をローカル実行すれば確かにAPIコストは低いですが、クラウドモデルを使うタスクのコストは馬鹿になりません。特にEnterpriseプランの「月無制限」は、コスト試算なしに設定すると赤字になります。
def calculate_plan_floor_price(
avg_tasks_per_month: int,
avg_tokens_per_task: int,
cloud_task_ratio: float = 0.3 # 30%はクラウドモデル使用を想定
) -> int:
"""プランの最低価格をAPIコストから逆算する"""
cloud_tasks = avg_tasks_per_month * cloud_task_ratio
tokens = cloud_tasks * avg_tokens_per_task
api_cost_jpy = (tokens / 1_000_000) * 15.0 * 150 # $15/1M tokens * ¥150
# 運営コスト・利益を含めた最低価格（APIコストの5倍）
floor_price = int(api_cost_jpy * 5)
return floor_price
floor = calculate_plan_floor_price(3000, 8000)
print(f"Growthプランの最低価格: ¥{floor:,}") # → ¥81,000程度ミス2: オンボーディングが不十分で早期解約が多発する
エージェントサービスの解約の60%以上は最初の1ヶ月で起きます。エージェントが「自分たちの業務を理解している」と感じてもらうまでの期間が最も離脱リスクが高い。14日の試用期間中に「ここまでできる」という体験を作ることが最優先です。
ミス3: ログとデバッグが不十分でトラブル対応が遅い
# 本番稼働に必須のロギング設定
import structlog
log = structlog.get_logger()
async def execute_agent_task_with_audit(
tenant_id: str,
task: str,
agent: CustomerSuccessAgent
) -> dict:
"""監査ログ付きでエージェントタスクを実行する"""
execution_id = generate_execution_id()
log.info("agent_task_started",
tenant_id=tenant_id,
execution_id=execution_id,
task_preview=task[:100])
try:
result = await agent.process_task(task)
log.info("agent_task_completed",
tenant_id=tenant_id,
execution_id=execution_id,
tokens_used=result["cost_tokens"],
duration_ms=result.get("duration_ms"))
return result
except Exception as e:
log.error("agent_task_failed",
tenant_id=tenant_id,
execution_id=execution_id,
error=str(e),
exc_info=True)
# 顧客へのエスカレーション
await notify_cs_team(tenant_id, execution_id, str(e))
raise6ヶ月でMRR200万円を目指すロードマップ
Month 1: 検証フェーズ（目標: MRR ¥0→¥30万）
- 1つの業種・1つのユースケースに絞る
- 無料デモ → 有料化（Starterプラン ¥30,000/月）
- 5クライアントの獲得
Month 2〜3: 拡大フェーズ（目標: MRR ¥80万）
- Starterから Growthへのアップグレード促進（¥30K→¥80K）
- 解約防止の仕組みを実装（月次レポート・チャーン検知）
- 横展開（同業種の紹介獲得）
Month 4〜5: スケールフェーズ（目標: MRR ¥150万）
- Enterpriseプランの提案開始（¥200K/月）
- パートナー（業種コンサルタント）経由の紹介獲得
- エージェント機能の拡張（追加ユースケース）
Month 6: 安定フェーズ（目標: MRR ¥200万）
- チャーン率5%以下の安定運用
- チームへの委任（CS担当・サポート）
- 次の業種への横展開計画
AgentKit 2.0の永続メモリとコスト管理APIは、このスケールアップを技術的に支えてくれます。まずは1つの業種で「エージェントが欠かせない」と思ってもらえる体験を作ることから始めてください。
初月5クライアントというのは、知人・コミュニティ・SNSから十分に現実的な数字です。今週末に1つ、身近な業務課題のデモを作ることから始めてみてください。