この記事のゴール
Backlog Webhook → AWS Lambda → Claude (Anthropic Sonnet 4) → Backlog という最小構成（MVP）で、「課題コメントに @claude と書く→Bot が自動返信」という仕組みを動かします。
IP ホワイトリスト & クエリパラメータトークンで “それなり” の認証も入れます。
1. アーキテクチャ概要
┏━━━━━━━━━━━━┓ Webhook(JSON) ┏━━━━━━━━━━━┓ Bedrock Invoke ┏━━━━━━━━━━━┓
┃ Backlog ┣━━━━━━━━━━━━━━▶┃ AWS Lambda ┣━━━━━━━━━━━━━━▶┃ Claude ┃
┃ (SaaS) ┃ ┃ (Python) ┃ コメントPOST ┃ Sonnet‑4 ┃
┗━━━━━━━━━━━━┛ ◀──────────────┗━━━━━━━━━━━┛ ◀──────────────┗━━━━━━━━━━━┛
▲ ▲
│ │
│ │
Backlog API IP & Token 認証
-
MVP: Lambda 関数 URL を直接叩く構成
（大量トラフィックをさばきたい場合は API Gateway＋WAF＋VPC‑ENI などに置き換え可） -
認証
- 環境変数
ALLOWED_IPSによる IP ホワイトリスト - Webhook 送信側で付与する
?token=...をBACKLOG_WEBHOOK_TOKENと照合
- 環境変数
2. 事前準備（Backlog 側）
| 手順 | 内容 | 状態 |
|---|---|---|
| 1 | Bot 用ユーザーを作成 | ✔ |
| 2 | Bot ユーザーの 個人 API キー取得 | ✔ |
| 3 | Webhook を作成（イベント: コメント追加）送信先 URL は後述の Lambda 関数 URL＋?token=YOUR_TOKEN
|
✔ |
| 4 | Webhook の テストボタンでペイロードを確認 | ✔ |
Memo : Webhook サンプル JSON は CloudWatch に全量ログ出力させて構造を把握しておくと後々が楽です。
3. 事前準備（AWS 側）
| 手順 | 内容 | ポイント |
|---|---|---|
| 1 | Bedrock の Claude‑Sonnet 4 を使用できるリージョンで有効化 | ap‑northeast‑1 など |
| 2 | IAM: Lambda 実行ロールに bedrock:InvokeModel 権限を付与 |
|
| 3 | Lambda 関数を Python 3.12 で作成 | コンテナ/Layers 不要 |
| 4 | 環境変数を設定 |
BACKLOG_BASE_URL, BACKLOG_API_KEY, ALLOWED_IPS, BACKLOG_WEBHOOK_TOKEN, ほか |
| 5 | 関数 URL を発行（Auth＝NONE） | 生成された HTTPS URL を Webhook に登録 |
4. Lambda 実装 ― 要所コード & 解説
フルソース（約 350 行）は 次のリンク先にあります。
ここでは “要所” を抜粋しながら動きのポイントを示します。
4‑1. ロガー & 環境変数
# -*- coding: utf-8 -*-
import json, logging, os, re, urllib.parse, urllib.request
from typing import Any, Dict, List, Tuple
import boto3
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=LOG_LEVEL,
format="%(levelname)s\t%(asctime)s\t%(name)s\t%(message)s",
force=True)
logger = logging.getLogger(__name__)
REGION = os.getenv("BEDROCK_REGION", "ap-northeast-1")
MODEL_ID = os.getenv("BEDROCK_MODEL_ID",
"apac.anthropic.claude-sonnet-4-20250514-v1:0")
MAX_TOKENS = int(os.getenv("MAX_TOKENS", "300"))
TEMPERATURE = float(os.getenv("TEMPERATURE", "0.3"))
BACKLOG_BASE_URL = os.getenv("BACKLOG_BASE_URL") # 例: https://xxx.backlog.com
BACKLOG_API_KEY = os.getenv("BACKLOG_API_KEY") # Bot ユーザーの個人 API キー
ALLOWED_IPS = {ip for ip in os.getenv("ALLOWED_IPS", "").split("|") if ip}
BACKLOG_WEBHOOK_TOKEN = os.getenv("BACKLOG_WEBHOOK_TOKEN", "").strip()
br = boto3.client("bedrock-runtime", region_name=REGION)
- 環境変数に寄せ切ることで IaC なしでも デプロイ先を簡単に差し替え可能。
-
force=Trueで 再デプロイ時のハンドラ再読み込みでもロガー設定が上書きされます。
4‑2. IP & Token 認証ユーティリティ
def source_ip(evt: Dict[str, Any]) -> str | None:
rc = evt.get("requestContext", {})
return rc.get("http", {}).get("sourceIp") or rc.get("identity", {}).get("sourceIp")
def ip_allowed(ip: str | None) -> bool:
# ALLOWED_IPS が空ならホワイトリストを無効化
return not ALLOWED_IPS or ip in ALLOWED_IPS
def token_valid(evt: Dict[str, Any]) -> bool:
# BACKLOG_WEBHOOK_TOKEN が空ならトークン認証を無効化
if not BACKLOG_WEBHOOK_TOKEN:
return True
qs = evt.get("queryStringParameters") or {}
return qs.get("token") == BACKLOG_WEBHOOK_TOKEN
- 同期 Webhook のため、冒頭で即 403 を返す のがベストプラクティス。
- ホワイトリストもトークンも 環境変数を空にすれば無効化 できるため、開発環境と本番を切り替えやすい。
4‑3. Backlog API ラッパ & フォールバック方針
def backlog_get(path: str, params: Dict[str, str] | None = None) -> Any:
params = {**(params or {}), "apiKey": BACKLOG_API_KEY}
url = f"{BACKLOG_BASE_URL}{path}?{urllib.parse.urlencode(params)}"
with urllib.request.urlopen(url, timeout=10) as resp:
if resp.status != 200:
raise RuntimeError(f"Backlog API status={resp.status}")
return json.load(resp)
def backlog_post_comment(issue_id: str, content: str) -> None:
url = (f"{BACKLOG_BASE_URL}/api/v2/issues/{issue_id}/comments"
f"?apiKey={BACKLOG_API_KEY}")
data = urllib.parse.urlencode({"content": content}).encode()
urllib.request.urlopen(urllib.request.Request(url, data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
method="POST"), timeout=10)
-
Backlog の Add Comment エンドポイント:
/api/v2/issues/:issueIdOrKey/comments([Nulab Developer API][1]) -
GET ２本（課題本体／コメント一覧）が 失敗した場合は Webhook 本体に入っている “stub” 情報にフォールバック。
停止より劣化稼働 を優先します。
4‑4. Invoke Claude (Bedrock)
def invoke_claude(prompt: str) -> str:
resp = br.invoke_model(
modelId=MODEL_ID,
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": MAX_TOKENS,
"temperature": TEMPERATURE,
}),
accept="application/json",
contentType="application/json",
)
return json.loads(resp["body"].read())["content"][0]["text"]
-
bedrock:InvokeModel1 回で完結。 - Claude 用のパラメータは Anthropic API と同一なのでモデル切替も容易。
- Bedrock InvokeModel の公式リファレンスはこちら ([AWS Documentation][2], [AWS Documentation][3])
4‑5. Webhook 解析 & プロンプト整形
MENTION_PATTERN = re.compile(r"@claude\b", re.IGNORECASE)
def parse_webhook(evt: Dict[str, Any]) -> Tuple[str, Dict[str, Any], Dict[str, Any]]:
body = json.loads(evt.get("body", "{}"))
content = body["content"]
issue_id = content.get("id") or content.get("key_id")
return str(issue_id), content, content["comment"]
def build_prompt(issue: Dict[str, Any],
comments: List[Dict[str, Any]],
trigger: Dict[str, Any]) -> str:
meta = f"""
[課題キー] {issue.get('issueKey', issue.get('id'))}
[タイトル] {issue.get('summary','')}
[状態] {issue.get('status', {}).get('name','')}
[担当者] {issue.get('assignee', {}).get('name','未割当')}
[説明]
{issue.get('description','').strip()}
""".strip()
history = "\n".join(
f"◦ {c.get('createdUser', {}).get('name','?')}: {c.get('content','')}"
for c in comments)
return (
f"<issue_info>{meta}</issue_info>\n"
f"<comments_history>{history}</comments_history>\n"
f"<trigger_comment>{trigger.get('content','')}</trigger_comment>\n"
f"<instruction>あなたは Backlog Bot です。"
f"<trigger_comment> に日本語で丁寧に回答してください。</instruction>"
)
- “XML 風タグ” で Claude に 構造付きコンテキストを渡すと一撃で高精度。
-
MENTION_PATTERNで @claude が含まれないコメントは即スキップしコストを節約。
4‑6. lambda_handler（骨格）
def lambda_handler(event: Dict[str, Any], context):
logger.info("Raw event: %s", json.dumps(event)[:4000])
# --- 認証 ---
if not ip_allowed(source_ip(event)) or not token_valid(event):
return {"statusCode": 403, "body": "Forbidden"}
try:
issue_id, issue_stub, trigger = parse_webhook(event)
if not MENTION_PATTERN.search(trigger.get("content", "")):
return {"statusCode": 200, "body": json.dumps({"message": "ignored"})}
# --- Backlog から課題・履歴取得（フォールバックあり） ---
try:
issue = backlog_get(f"/api/v2/issues/{issue_id}") # :contentReference[oaicite:2]{index=2}
except Exception:
issue = issue_stub
try:
comments = backlog_get(f"/api/v2/issues/{issue_id}/comments",
{"count": "100", "order": "asc"})
except Exception:
comments = []
# --- Claude 呼び出し & コメント投稿 ---
answer = invoke_claude(build_prompt(issue, comments, trigger))
backlog_post_comment(issue_id, answer)
return {"statusCode": 200, "body": json.dumps({"message": "posted"})}
except Exception as e:
logger.exception("Unhandled error")
return {"statusCode": 500, "body": str(e)}
-
処理時間を短縮するため、Bedrock への呼び出し前に不要な取得を省く（
MENTION_PATTERNでフィルタ）。 - 失敗時は CloudWatch Logs に詳細を残すので
LOG_LEVEL=DEBUGを一時的に設定してトラブルシュート。
5. 仕組みの流れ（おさらい）
| 箇所 | 何をしているか | 補足 |
|---|---|---|
| IP & Token 認証 |
ip_allowed()／token_valid()
|
同期 Webhook なので先頭で 403 を返す |
| Webhook 解析 | parse_webhook() |
issue_id・issue_stub・trigger_comment を抽出 |
| 課題 & コメント取得 | Backlog API GET /issues/:id と /comments
|
失敗時は Webhook 情報にフォールバック |
| プロンプト整形 |
<issue_info> ほか “XML 風タグ” |
Claude に一発で大量コンテキストを渡す |
| Claude 生成 | invoke_claude() |
Bedrock InvokeModel 1 回で完結 |
| 自動返信投稿 | backlog_post_comment() |
すぐ Backlog に反映 |
6. デプロイ & テスト
-
Backlog で任意の課題にコメント例:
@claude 「API キーの取得手順を教えて」 -
2–3 秒後、Bot が同スレッドに回答を自動投稿
-
CloudWatch Logs で prompt / Claude 応答 / POST 結果 を確認
トラブル時はLOG_LEVEL=DEBUG推奨
7. 運用 TIPS / 今後の改良案
| 課題 | MVP 対処 | スケール時の解決策 |
|---|---|---|
| 同時呼び出し数制限 | Lambda の同時実行≒並列回答 | Step Functions＋SQS で平滑化 |
| IP ホワイトリスト管理 | 環境変数に直書き | VPC＋Private Link で閉域に |
| 無制限の関数 URL | 固定トークンで最低限の認証 | API Gateway＋WAF＋Auth |
| 長大コメントで prompt オーバー |
MAX_TOKENS を絞る |
Claude Stream＋分割再構築 |
| 監査ログ | CloudWatch のみ | EventBridge → S3 へ集約 |
8. まとめ
- Webhook 1 行の @claude に反応して即回答 —— 🤖💬 思考ログをチームに残せる
- IP & Token 二段構えで “とりあえず” の安全を確保
- 完全サーバーレス: メンテは 環境変数とコード だけ
- 伸びしろは API Gateway／SQS／Step Functions／VPC‑ENI など、要求に応じて後付け可能
📌 Fork 自由 — ぜひ自社環境に合わせて改造し、Bedrock + Backlog の威力を体感してください！