Claude APIでAIチャットボットを実装する

出来合いのボットを諦めて、自分で組み直した夜

最初に作った問い合わせ用のボットは、出来合いのサービスの上に乗せていました。動くには動くのですが、返答の言い回しをひと言だけ直したいときに、触れる場所が見つかりません。ある夜、管理画面のタブを開いては閉じるのを三十分ほど繰り返して、そこで手が止まりました。

いま思えば、私が欲しかったのは高機能なボットではなく、中身を自分で開けられるボットだったのだと思います。Claude API を直接叩く形に組み直してからは、システムプロンプトの一字も、履歴をどこで切るかも、どのモデルにいくら払うかも、自分の判断で決められるようになりました。

以下は、そのとき私がたどった順番をそのまま並べたものです。最小構成から始めて、履歴・ストリーミング・Web API・コスト・リトライ・PII まで、動くコードで一段ずつ積み上げてまいります。

ℹ️

本記事のコードはすべて **Python 3.10以上** で動作確認済みです。APIキーの取得については [Claude API クイックスタート](/articles/api-sdk/api-quickstart) を参照してください。

STEP 1：最小構成のチャットボット

まず、最もシンプルな実装から始めましょう。

import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
def chat(user_message: str) -> str:
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": user_message}
        ]
    )
    return message.content[0].text
 
# 実行
response = chat("Pythonで素数を判定する関数を書いてください")
print(response)

これだけで動きます。しかし会話の文脈が保持されないという致命的な問題があります。次のメッセージを送ると、前の会話を覚えていません。

STEP 2：会話履歴を管理する

Claude APIはステートレスなので、会話履歴をクライアント側で管理する必要があります。

import anthropic
from typing import List
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
class ChatSession:
    def __init__(self, system_prompt: str = ""):
        self.history: List[dict] = []
        self.system_prompt = system_prompt
 
    def send(self, user_message: str) -> str:
        # ユーザーメッセージを履歴に追加
        self.history.append({
            "role": "user",
            "content": user_message
        })
 
        # API呼び出し（システムプロンプト + 全履歴を送信）
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=self.system_prompt,
            messages=self.history
        )
 
        assistant_message = response.content[0].text
 
        # アシスタントの返答を履歴に追加
        self.history.append({
            "role": "assistant",
            "content": assistant_message
        })
 
        return assistant_message
 
    def clear(self):
        """会話をリセット"""
        self.history = []
 
# 使用例
session = ChatSession(
    system_prompt="あなたはPython専門のコードレビュアーです。"
                  "コードの問題点を指摘し、改善案を提示してください。"
)
 
print(session.send("このコードをレビューしてください：\ndef add(a, b): return a+b"))
print(session.send("では、型ヒントを追加したバージョンも書いてください"))  # 前の文脈を覚えている

ℹ️

`system` パラメータは会話履歴に含めず、毎回APIに渡すことで、どのターンでも一貫した振る舞いを保てます。

STEP 3：ストリーミング応答を実装する

長い応答を待たせると UX が悪化します。ストリーミングで文字を順次表示しましょう。

import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
class StreamingChatSession:
    def __init__(self, system_prompt: str = ""):
        self.history = []
        self.system_prompt = system_prompt
 
    def send_stream(self, user_message: str):
        """ストリーミングでテキストをyieldする"""
        self.history.append({"role": "user", "content": user_message})
 
        full_response = ""
 
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                yield text  # テキストを逐次返す
 
        # 完全な応答を履歴に保存
        self.history.append({"role": "assistant", "content": full_response})
 
# CLIでの使用例
session = StreamingChatSession(system_prompt="あなたは親切なアシスタントです。")
 
while True:
    user_input = input("\nYou: ").strip()
    if user_input.lower() in ["quit", "exit", "終了"]:
        break
 
    print("Claude: ", end="", flush=True)
    for chunk in session.send_stream(user_input):
        print(chunk, end="", flush=True)
    print()  # 改行

STEP 4：FastAPIでWeb APIとして公開する

CLIだけでなく、WebアプリからもアクセスできるAPIとして実装します。

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import json
from typing import Optional
 
app = FastAPI()
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# シンプルなインメモリセッション管理（本番ではRedis推奨）
sessions: dict[str, list] = {}
 
class ChatRequest(BaseModel):
    session_id: str
    message: str
    system_prompt: Optional[str] = "あなたは親切なアシスタントです。"
 
class ChatResponse(BaseModel):
    response: str
    session_id: str
 
@app.post("/chat")
async def chat(request: ChatRequest):
    """通常のチャットエンドポイント"""
    if request.session_id not in sessions:
        sessions[request.session_id] = []
 
    history = sessions[request.session_id]
    history.append({"role": "user", "content": request.message})
 
    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=request.system_prompt,
            messages=history
        )
 
        assistant_message = response.content[0].text
        history.append({"role": "assistant", "content": assistant_message})
 
        return ChatResponse(
            response=assistant_message,
            session_id=request.session_id
        )
    except anthropic.APIError as e:
        raise HTTPException(status_code=500, detail=str(e))
 
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """Server-Sent Events でストリーミング"""
    if request.session_id not in sessions:
        sessions[request.session_id] = []
 
    history = sessions[request.session_id]
    history.append({"role": "user", "content": request.message})
 
    def generate():
        full_response = ""
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=request.system_prompt,
            messages=history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                # SSE形式で送信
                yield f"data: {json.dumps({'text': text})}\n\n"
 
        history.append({"role": "assistant", "content": full_response})
        yield f"data: {json.dumps({'done': True})}\n\n"
 
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )
 
@app.delete("/session/{session_id}")
async def clear_session(session_id: str):
    """セッションをリセット"""
    sessions.pop(session_id, None)
    return {"status": "cleared"}

起動方法：

pip install fastapi uvicorn anthropic
uvicorn main:app --reload

STEP 5：会話履歴のコスト最適化

会話が長くなるほどトークン消費が増大します。主要な最適化テクニックを紹介します。

5-1：会話を要約してトリミング

def summarize_and_trim(
    client: anthropic.Anthropic,
    history: list,
    max_turns: int = 10
) -> list:
    """古い会話を要約して履歴を圧縮する"""
    if len(history) <= max_turns * 2:
        return history
 
    # 古い部分を要約
    old_history = history[:-max_turns * 2]
    recent_history = history[-max_turns * 2:]
 
    summary_response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 安価なモデルで要約
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"以下の会話を3文以内で要約してください：\n\n{json.dumps(old_history, ensure_ascii=False)}"
        }]
    )
 
    summary = summary_response.content[0].text
 
    # 要約をシステムメッセージとして先頭に挿入
    return [
        {"role": "user", "content": f"[会話の要約] {summary}"},
        {"role": "assistant", "content": "了解しました。前の会話の文脈を把握しています。"},
        *recent_history
    ]

5-2：Prompt Caching でAPI料金を削減

システムプロンプトが長い場合、キャッシュで最大90%のコスト削減が可能です。

# キャッシュを活用したシステムプロンプト
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "ここに長いシステムプロンプト（社内ドキュメントや仕様書など）...",
            "cache_control": {"type": "ephemeral"}  # キャッシュを有効化
        }
    ],
    messages=history
)

ℹ️

Prompt Caching の詳細は [Claude API のプロンプトキャッシュで月額コストを半分にした実装メモ](/articles/api-sdk/claude-api-prompt-caching-monthly-cost-half-guide) をご覧ください。1000トークン以上のシステムプロンプトで効果が出ます。

5-3：用途に応じてモデルを使い分ける

ユースケース	推奨モデル	理由
単純なQ&A	claude-haiku-4-5-20251001	高速・安価
コーディング支援	claude-sonnet-4-6	バランス良好
複雑な分析・推論	claude-opus-4-6	最高精度
要約・分類	claude-haiku-4-5-20251001	コスト最適

def get_model_for_task(task_type: str) -> str:
    """タスクに応じてモデルを自動選択"""
    model_map = {
        "simple_qa": "claude-haiku-4-5-20251001",
        "coding": "claude-sonnet-4-6",
        "analysis": "claude-opus-4-6",
        "summarization": "claude-haiku-4-5-20251001",
    }
    return model_map.get(task_type, "claude-sonnet-4-6")

STEP 6：エラーハンドリングとリトライ — 待ち方をそろえない

ここは私が一度作り直した箇所です。最初は下のように、2 ** attempt で待ち時間を伸ばすだけの素直な指数バックオフを書いておりました。

import anthropic
import time
from typing import Optional
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
def safe_chat(
    messages: list,
    system: str = "",
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> Optional[str]:
    """リトライ付きのAPI呼び出し"""
 
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=2048,
                system=system,
                messages=messages
            )
            return response.content[0].text
 
        except anthropic.RateLimitError:
            # レート制限：指数バックオフで待機
            wait_time = retry_delay * (2 ** attempt)
            print(f"レート制限に達しました。{wait_time}秒後にリトライ...")
            time.sleep(wait_time)
 
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:
                # サーバーエラー：リトライ
                print(f"サーバーエラー ({e.status_code})。リトライ中...")
                time.sleep(retry_delay)
            else:
                # クライアントエラー（400番台）：リトライしない
                print(f"クライアントエラー: {e.message}")
                return None
 
        except anthropic.APIConnectionError:
            # 接続エラー：リトライ
            print("接続エラー。リトライ中...")
            time.sleep(retry_delay)
 
    print("最大リトライ回数に達しました")
    return None

これで動いてはいたのですが、利用者が増えた時間帯にだけ、失敗がまとまって出ました。原因を追ってみると、コードではなく待ち方のほうに問題がありました。全員が同じ式で待つと、全員が同じ瞬間に戻ってきます。 1 秒後、2 秒後、4 秒後——足並みがそろったまま、同じ壁に何度もぶつかっていたのです。

待ち方の違いを手元で測ってみる

腑に落ちなかったので、API を呼ばずに待ち方だけを取り出して数えてみました。50 個のクライアントが同時に 429 を受け、サーバー側は 1 秒あたり 5 件しか通せない、リトライ上限は 5 回、という条件です。乱数の種を変えた 20 回の平均が下の値になります。

待ち方	1 秒窓の最大同時到着数	API 呼び出し総数	打ち切り	成功までの中央値
`2 ** attempt`（ジッタなし）	50.0	275.0	20.0 件	11.0 秒
`uniform(0, 2 ** attempt)`（full jitter）	8.3	107.5	0 件	0.6 秒

同じ回数だけ待っているのに、結果はここまで離れます。ジッタなしでは 50 件が一斉に戻ってくるため定員 5 件からあふれ続け、20 件が上限に達して落ちました。乱数で散らすと山が 8 件ほどに崩れ、呼び出し総数も 4 割に減って、打ち切りはゼロになっています。待ち時間を伸ばすことより、戻る時刻をばらけさせることのほうが効いていた、というのがこの表から読み取れたことでした。

再現用のスクリプトは短いので、ご自身の同時実行数と定員に置き換えて回してみていただければと思います。

import random, statistics
 
N_CLIENTS, MAX_RETRIES, BASE, CAPACITY = 50, 5, 1.0, 5
 
def wait(strategy, attempt, rng):
    if strategy == "plain":
        return round(BASE * (2 ** attempt), 1)              # 全員が同じ時刻に戻る
    return round(rng.uniform(0, BASE * (2 ** attempt)), 1)  # full jitter
 
def run(strategy, seed):
    rng = random.Random(seed)
    cs = [{"attempt": 0, "next_t": wait(strategy, 0, rng), "done": False}
          for _ in range(N_CLIENTS)]
    t, peak, calls, ok = 0.0, 0, N_CLIENTS, []
    while t < 300 and any(not c["done"] for c in cs):
        arrivals = [c for c in cs if not c["done"] and abs(c["next_t"] - t) < 1e-9]
        rng.shuffle(arrivals)
        peak = max(peak, len(arrivals))
        accepted = 0
        for c in arrivals:
            calls += 1
            if accepted < CAPACITY:                 # 定員内なら通る
                accepted += 1
                c["done"] = True
                ok.append(t)
            else:
                c["attempt"] += 1
                if c["attempt"] > MAX_RETRIES:
                    c["done"] = True                # 打ち切り
                else:
                    c["next_t"] = round(t + wait(strategy, c["attempt"], rng), 1)
        t = round(t + 0.1, 1)
    return peak, calls, len(ok), statistics.median(ok)
 
for s in ("plain", "full_jitter"):
    r = [run(s, seed) for seed in range(2000, 2020)]
    print(s, [round(statistics.mean(col), 1) for col in zip(*r)])

サーバーが教えてくれる待ち時間を先に読む

もう一つ、あとから知って作り直した点があります。Claude API は 429 を返すときに retry-after ヘッダーを添えてくれます。こちらで推測した秒数より、そちらのほうが正確です。

さらに注意したいのが、429 のすべてに retry-after が付くわけではないという点です。利用枠の上限に達したことによる 429 にはこのヘッダーが付かず、待って再送しても状況は変わりません。ヘッダーの有無で「待てば通るのか、待っても通らないのか」を分けられます（Rate limits — Claude Platform Docs）。

この二つを反映すると、リトライ部分は次の形に落ち着きました。

import random
import time
from typing import Optional
 
import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
 
def _retry_after_seconds(err) -> Optional[float]:
    # サーバーが待ち時間を指定していれば、その秒数を返す
    headers = getattr(getattr(err, "response", None), "headers", None) or {}
    raw = headers.get("retry-after")
    if raw is None:
        return None
    try:
        return float(raw)
    except (TypeError, ValueError):
        return None
 
 
def safe_chat(
    messages: list,
    system: str = "",
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
) -> Optional[str]:
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=2048,
                system=system,
                messages=messages,
            )
            return response.content[0].text
 
        except anthropic.RateLimitError as e:
            hinted = _retry_after_seconds(e)
            if hinted is None and attempt >= 1:
                # 枠の上限による 429 は待っても解消しないため、早めに呼び出し元へ返す
                print("retry-after が付かない 429 です。利用枠の上限をご確認ください")
                return None
            ceiling = min(hinted or base_delay * (2 ** attempt), max_delay)
            wait_time = random.uniform(0, ceiling)  # full jitter
            print(f"レート制限です。{wait_time:.1f} 秒後に再送します")
            time.sleep(wait_time)
 
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:
                wait_time = random.uniform(0, min(base_delay * (2 ** attempt), max_delay))
                print(f"サーバーエラー ({e.status_code})。{wait_time:.1f} 秒後に再送します")
                time.sleep(wait_time)
            else:
                # 400 番台は何度送っても同じ結果になるため、ここで返す
                print(f"クライアントエラー: {e.message}")
                return None
 
        except anthropic.APIConnectionError:
            wait_time = random.uniform(0, min(base_delay * (2 ** attempt), max_delay))
            print(f"接続エラー。{wait_time:.1f} 秒後に再送します")
            time.sleep(wait_time)
 
    print("最大リトライ回数に達しました")
    return None

変えたのは三点だけです。サーバーの指定を優先すること、待ち時間に乱数を掛けること、そして待っても通らない 429 を区別して早く手を引くこと。再送で守るのは自分のリクエストではなく、次の 1 秒に並ぶ全員の順番です。 この考え方に切り替えてから、混雑する時間帯の失敗の出方が変わりました。

STEP 7：完成形 — 実用的なチャットボットクラス

これまでの要素をまとめた、本番投入できるクラスです。

import anthropic
import json
import time
import uuid
from typing import Generator, Optional
 
class ProductionChatBot:
    """本番用チャットボットクラス"""
 
    def __init__(
        self,
        api_key: str,
        system_prompt: str = "あなたは親切なアシスタントです。",
        model: str = "claude-sonnet-4-6",
        max_tokens: int = 2048,
        max_history_turns: int = 20,
    ):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.system_prompt = system_prompt
        self.model = model
        self.max_tokens = max_tokens
        self.max_history_turns = max_history_turns
        self.history = []
        self.session_id = str(uuid.uuid4())
        self.total_tokens_used = 0
 
    def send(self, message: str) -> str:
        """メッセージを送信してレスポンスを返す"""
        self.history.append({"role": "user", "content": message})
 
        response = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            system=self.system_prompt,
            messages=self.history
        )
 
        assistant_text = response.content[0].text
        self.history.append({"role": "assistant", "content": assistant_text})
 
        # トークン使用量を記録
        self.total_tokens_used += response.usage.input_tokens + response.usage.output_tokens
 
        # 履歴が長くなりすぎたらトリミング
        if len(self.history) > self.max_history_turns * 2:
            self.history = self.history[-(self.max_history_turns * 2):]
 
        return assistant_text
 
    def stream(self, message: str) -> Generator[str, None, None]:
        """ストリーミングでテキストをyield"""
        self.history.append({"role": "user", "content": message})
        full_response = ""
 
        with self.client.messages.stream(
            model=self.model,
            max_tokens=self.max_tokens,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                yield text
 
        self.history.append({"role": "assistant", "content": full_response})
 
    def get_stats(self) -> dict:
        """使用統計を返す"""
        return {
            "session_id": self.session_id,
            "turns": len(self.history) // 2,
            "total_tokens": self.total_tokens_used,
            "estimated_cost_usd": self.total_tokens_used / 1_000_000 * 3.0  # Sonnet概算
        }
 
    def reset(self):
        """会話をリセット"""
        self.history = []
        self.total_tokens_used = 0
 
# 使用例
bot = ProductionChatBot(
    api_key="YOUR_API_KEY",
    system_prompt="""あなたはプロのPython開発者です。
    コードの品質、可読性、パフォーマンスに強いこだわりを持ち、
    ベストプラクティスに基づいた実装を提案してください。"""
)
 
# ストリーミング応答
print("Claude: ", end="")
for chunk in bot.stream("FastAPIでCRUDアプリを作るベストプラクティスを教えてください"):
    print(chunk, end="", flush=True)
print()
 
print(f"\n📊 統計: {bot.get_stats()}")

STEP 8：LINE・Telegram・Discord に接続する — プラットフォーム固有の実装差分

これまではターミナルや自分の Web API でチャットボットを動かしてきました。実際に人に使ってもらうとなると、LINE・Telegram・Discord のようなメッセージングプラットフォームに載せる場面が多いはずです。

私自身、個人開発で作ったチャットボットを最初に LINE へ載せたとき、いちばん面食らったのが reply token の寿命でした。

どちらも「プラットフォームが Webhook であなたのサーバーに POST する → 署名を検証する → Claude に投げる → 返信する」という骨格は同じです。違いはその細部に出ます。ここでつまずきやすい差分を整理しておきます。

LINE：reply token の寿命が落とし穴

LINE Messaging API で最初にハマるのが、reply token の扱いです。reply token は 1 回しか使えず、有効期限も短いのです（発行から 1 分ほど）。Claude の応答に数秒かかる程度なら間に合いますが、長文生成やツール呼び出しを挟むと失効してしまうことがあります。

実装上は「まず loading アニメーションを表示して reply token を温存し、応答が固まってから push message で送る」という二段構えが安定します。

署名検証も必須です。X-Line-Signature ヘッダーを検証しないと、誰でもあなたのエンドポイントに偽のイベントを送れてしまいます。

import hmac
import hashlib
import base64
import os
import requests
from fastapi import FastAPI, Request, HTTPException
 
app = FastAPI()
LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
LINE_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
 
def verify_line_signature(body: bytes, signature: str) -> bool:
    """X-Line-Signature を HMAC-SHA256 で検証する"""
    digest = hmac.new(
        LINE_CHANNEL_SECRET.encode("utf-8"), body, hashlib.sha256
    ).digest()
    expected = base64.b64encode(digest).decode("utf-8")
    return hmac.compare_digest(expected, signature)
 
def line_push(user_id: str, text: str) -> None:
    """reply token に依存しない push message で送る"""
    requests.post(
        "https://api.line.me/v2/bot/message/push",
        headers={"Authorization": f"Bearer {LINE_ACCESS_TOKEN}"},
        json={"to": user_id, "messages": [{"type": "text", "text": text}]},
        timeout=10,
    )
 
@app.post("/line/webhook")
async def line_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Line-Signature", "")
    if not verify_line_signature(body, signature):
        raise HTTPException(status_code=403, detail="signature mismatch")
 
    events = (await request.json()).get("events", [])
    for event in events:
        if event.get("type") == "message" and event["message"]["type"] == "text":
            user_id = event["source"]["userId"]
            user_text = event["message"]["text"]
            # ユーザーごとに履歴を持つ（STEP 7 の ProductionChatBot を流用）
            reply = get_bot_for(user_id).send(user_text)
            line_push(user_id, reply)
    return {"ok": True}

get_bot_for(user_id) は、ユーザー ID をキーに ProductionChatBot のインスタンスを引く想定です。プロセス内の辞書で持つと再起動で消えるので、本番では会話履歴を Redis などの外部ストアに逃がしておくと安心です。

Telegram：webhook か long polling かを使い分ける

Telegram には受信方法が 2 通りあります。getUpdates をループで叩く long polling と、公開エンドポイントに POST させる webhook です。

開発中は polling のほうがはるかに楽です。公開 URL も HTTPS 証明書も要らず、ローカルでそのまま動きます。一方、本番でサーバーを常時起動しておくなら webhook のほうが無駄がありません。webhook を使うときは、登録時に secret_token を指定し、受信側で X-Telegram-Bot-Api-Secret-Token ヘッダーを突き合わせて検証します。

応答に時間がかかる Claude では、sendChatAction で「入力中…」を出しておくと体感がぐっと良くなります。

import os
import requests
 
TG_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
API = f"https://api.telegram.org/bot{TG_TOKEN}"
 
def tg_typing(chat_id: int) -> None:
    requests.post(f"{API}/sendChatAction",
                  json={"chat_id": chat_id, "action": "typing"}, timeout=10)
 
def tg_send(chat_id: int, text: str) -> None:
    requests.post(f"{API}/sendMessage",
                  json={"chat_id": chat_id, "text": text}, timeout=10)
 
def run_polling():
    """開発向け: long polling で受信する"""
    offset = None
    while True:
        resp = requests.get(f"{API}/getUpdates",
                            params={"timeout": 30, "offset": offset}, timeout=40)
        for update in resp.json().get("result", []):
            offset = update["update_id"] + 1
            msg = update.get("message")
            if not msg or "text" not in msg:
                continue
            chat_id = msg["chat"]["id"]
            tg_typing(chat_id)
            reply = get_bot_for(str(chat_id)).send(msg["text"])
            tg_send(chat_id, reply)

Discord：3 秒以内に「受け取った」と返す

LINE と Telegram はどちらも Webhook 方式でした。Discord も同じつもりで書き始めたところ、最初に作ったスラッシュコマンドがことごとく無反応で終わり、しばらく手が止まりました。

discord.py で組む場合、イベントは Webhook ではなく 常時接続の WebSocket（Gateway） 経由で届きます。署名検証は不要です。その代わりに別の締め切りがあります。スラッシュコマンドは 3 秒以内に何らかの応答を返さないと、その対話そのものが無効になります。

Claude の生成には数秒かかります。素直に「質問を受け取る → Claude を呼ぶ → 返信する」と書くと、ほぼ確実にこの 3 秒を踏み抜きます。

鍵は ctx.defer() です。最初にこれを呼んでおくと「処理中」の状態を Discord に伝えられ、実際の返信は followup.send() で後から送れます。LINE で reply token を温存してから push message に切り替える発想と、狙いはほとんど同じです。

もう一つの制約が 1 メッセージ 2000 文字 の上限です。Claude は放っておくと平気で超えてきますので、システムプロンプトで長さを促したうえで、送信側でも切り詰めておきます。

import discord
from discord import option
from discord.ext import commands
 
SYSTEM_PROMPT = """あなたは Discord サーバーに常駐する AI アシスタントです。
回答は簡潔にまとめ、1 メッセージ 2000 文字の上限を意識してください。
コードは必ずコードブロックで囲んでください。"""
 
class ClaudeCog(commands.Cog):
    def __init__(self, bot: commands.Bot):
        self.bot = bot
 
    @discord.slash_command(name="ask", description="Claude に質問します")
    @option("question", description="質問を入力してください")
    async def ask(self, ctx: discord.ApplicationContext, question: str):
        # 3 秒の締め切りをここで回避する。先に「処理中」を返しておく
        await ctx.defer()
 
        # STEP 7 の ProductionChatBot をユーザー単位で引く
        reply = get_bot_for(str(ctx.author.id)).send(question)
 
        # 2000 文字の上限に収める
        if len(reply) > 1900:
            reply = reply[:1900] + "\n\n…（長いため省略しました）"
 
        await ctx.followup.send(reply)
 
def setup(bot: commands.Bot):
    bot.add_cog(ClaudeCog(bot))

起動側では Message Content Intent を明示的に有効にします。コード側と Developer Portal 側の設定が両方揃っていないと、Bot は起動時点で例外を吐いて落ちます。

intents = discord.Intents.default()
intents.message_content = True  # Developer Portal 側でも ON にしておく
 
bot = commands.Bot(command_prefix="!", intents=intents)

Discord で私が実際に踏んだつまずきを、3 つだけ残しておきます。

症状	原因	対処
起動時に `PrivilegedIntentsRequired`	Developer Portal で Message Content Intent が OFF	Bot タブ → Privileged Gateway Intents → Message Content Intent を ON
スラッシュコマンドが候補に出てこない	招待 URL に `applications.commands` スコープが無い	OAuth2 URL Generator でスコープを追加し、招待し直す
コマンドを直したのに反映されない	グローバルコマンドは反映に最大 1 時間かかる	開発中はギルドコマンド（特定サーバー限定）で登録して即時反映させる

3 つ目は地味ですが、私はここで一番時間を溶かしました。コードが悪いと思い込んで書き直した先で、実はただ反映を待っていただけ、という徒労です。開発中はギルド限定で登録しておく——これだけで検証のテンポが変わります。

3 つのプラットフォームを並べて見る

こうして並べると、差分は「受信方式」「応答の締め切り」「返信の作法」の 3 点に収束します。

	LINE	Telegram	Discord
受信方式	Webhook	Webhook / long polling	Gateway（WebSocket 常時接続）
受信の検証	`X-Line-Signature` を HMAC-SHA256 で検証	`secret_token` ヘッダーを突き合わせ	不要（Gateway が認証済み）
応答の締め切り	reply token が約 1 分で失効	実質なし	スラッシュコマンドは 3 秒
締め切りの回避策	loading 表示 → push message	—	`ctx.defer()` → `followup.send()`
「入力中」表示	loading アニメーション API	`sendChatAction`	`channel.typing()`
1 通の文字数上限	5,000 文字	4,096 文字	2,000 文字

見比べていて腑に落ちたのは、どのプラットフォームも結局は同じことを要求している、という点でした。「あなたのサーバーが生きていることを、まず素早く知らせろ」。Claude の生成が終わるのを待ってから返事をしようとすると、名前を変えた同じ壁に必ずぶつかります。

先に「受け取りました」を返し、本文はあとから送ります。 この一行を最初から組んでおけば、載せ替え先がどこであれ、大きく崩れることはありません。

STEP 9：Claude に個人情報をそのまま渡さない — PII マスキング

個人開発の案件でも、問い合わせ対応をチャットボットに任せようとした途端にこの問題に直面しました。社内データや顧客とのやり取りをチャットボットに通すと、メールアドレスや電話番号といった個人情報（PII）が、そのまま API に流れてしまいます。入力ログにも、モデルへのリクエストにも生の値を残したくない、という要件は本番では珍しくありません。

有効なのは、送信前に PII をプレースホルダーへ置き換え、応答が返ってきてから元に戻すという可逆マスキングです。Claude には {{EMAIL_1}} のような無害化された文字列だけが渡り、ユーザーには復元済みの自然な文章が返ります。

import re
 
class PIIMasker:
    """送信前にPIIを伏字化し、応答後に復元する可逆マスカー"""
 
    PATTERNS = {
        "EMAIL": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
        "PHONE": re.compile(r"0\d{1,4}-\d{1,4}-\d{4}"),
        "CARD": re.compile(r"\b(?:\d[ -]?){13,16}\b"),
    }
 
    def mask(self, text: str):
        mapping = {}
        counters = {}
        def replace(kind, m):
            counters[kind] = counters.get(kind, 0) + 1
            token = f"{{{{{kind}_{counters[kind]}}}}}"
            mapping[token] = m.group(0)
            return token
        for kind, pattern in self.PATTERNS.items():
            text = pattern.sub(lambda m, k=kind: replace(k, m), text)
        return text, mapping
 
    def unmask(self, text: str, mapping: dict) -> str:
        for token, original in mapping.items():
            text = text.replace(token, original)
        return text
 
# 使用例
masker = PIIMasker()
user_text = "請求書を tanaka@example.com に送って、090-1234-5678 にも連絡して"
masked, mapping = masker.mask(user_text)
print(masked)
# → 請求書を {{EMAIL_1}} に送って、{{PHONE_1}} にも連絡して
 
reply = bot.send(masked)          # Claude には伏字だけが渡る
restored = masker.unmask(reply, mapping)  # ユーザーには元の値で返す

ひとつ正直に書いておきたいのは、正規表現によるマスキングは万能ではないということです。メールや電話番号のような形式が決まっているものは捕まえやすいですが、氏名や住所のように文脈で判断する PII は、固有表現抽出（NER）や辞書を併用しないと取りこぼします。

「絶対に外に出してはいけない情報」が決まっているなら、そもそもその項目を Claude に渡さない設計が最も確実です。マスキングは「うっかり混入を減らす一層の防御」と位置づけ、本当に守りたい情報は入力段階で切り離しておきます。この二段構えで考えておくと、後から監査が入っても説明できる基盤になります。

最初の一段だけ、今日のうちに

ここまで九段を積み上げてきましたが、全部を一度に組む必要はありません。私自身、最初の版は STEP 2 までで公開して、履歴の切り方に困ってから STEP 5 を、失敗がまとまって出てから STEP 6 を足しました。順番が逆でも、たぶん同じ場所でつまずいたのだと思います。

今日ひとつだけ手を動かすなら、STEP 6 の再送に random.uniform を一行入れるところをおすすめします。既存のボットにも数分で入りますし、混み合う時間帯の挙動が最初に変わるのがここでした。

Claude API の Parallel Tool Use を本番で使いこなす — ボットに検索や計算を持たせて、複数のツールを同時に走らせるときの設計
Claude Vision API 実装パターン — 画像や PDF を受け取れるボットへ広げるとき
Claude API のプロンプトキャッシュで月額コストを半分にした実装メモ — 会話が伸びて料金が気になってきたとき

出来合いのボットに戻ろうと思ったことは、あの夜以来ありません。中身が見えていれば、直したい一行に必ず手が届くからです。長い記事にお付き合いいただき、ありがとうございました。

Claude APIでAIチャットボットを実装する — ストリーミング・会話履歴・コスト最適化まで

出来合いのボットを諦めて、自分で組み直した夜

STEP 1：最小構成のチャットボット

STEP 2：会話履歴を管理する

STEP 3：ストリーミング応答を実装する

STEP 4：FastAPIでWeb APIとして公開する

STEP 5：会話履歴のコスト最適化

5-1：会話を要約してトリミング

5-2：Prompt Caching でAPI料金を削減

5-3：用途に応じてモデルを使い分ける

STEP 6：エラーハンドリングとリトライ — 待ち方をそろえない

待ち方の違いを手元で測ってみる

サーバーが教えてくれる待ち時間を先に読む

STEP 7：完成形 — 実用的なチャットボットクラス

STEP 8：LINE・Telegram・Discord に接続する — プラットフォーム固有の実装差分

LINE：reply token の寿命が落とし穴

Telegram：webhook か long polling かを使い分ける

Discord：3 秒以内に「受け取った」と返す

3 つのプラットフォームを並べて見る

STEP 9：Claude に個人情報をそのまま渡さない — PII マスキング

最初の一段だけ、今日のうちに

お読みいただきありがとうございます

関連記事