はじめに

前回までで、FastAPIを使った「認証付きREST API」が完成しました。

今回はいよいよシリーズの集大成です。

AnthropicのClaude APIをFastAPIでラップして、社内向けのAIツールの土台を作ります。

1. 「APIのラッパー」とは？

「ラッパー（wrapper）」とは「包むもの」という意味です。

外部のAPI（今回はClaude API）を自分のFastAPIサーバーで包むことで、こんなことができるようになります。

社内のユーザー
    ↓
自分のFastAPIサーバー（ラッパー）  ← ここで認証・ログ・制御をする
    ↓
Claude API（Anthropic）

なぜ直接Claude APIを呼ばないのか？

直接呼ぶと困ることがあります。

課題	ラッパーで解決できること
APIキーを全員に配れない	サーバー側だけで管理できる
誰がどれだけ使ったか見えない	ログを自分で取れる
会社のルールに沿ったプロンプトを強制したい	サーバー側で制御できる
使いすぎを防ぎたい	レート制限を自分で実装できる

2. 今回作るもの

シンプルな3つのエンドポイントを作ります。

POST /ai/chat          ── メッセージを送ってAIの返答を受け取る
POST /ai/chat/stream   ── ストリーミングで返答を受け取る（リアルタイム表示）
GET  /ai/usage/me      ── 自分の使用状況を確認する（認証必須）

前回までの認証（JWT）を組み合わせるので、ログインしたユーザーだけが使えるAI APIになります。

3. プロジェクト構成

前回の構成に ai.py を追加します。

my_api/
├── main.py
├── database.py
├── models.py      ← 使用ログ用テーブルを追加
├── schemas.py     ← AIリクエスト/レスポンスの型を追加
├── crud.py
├── auth.py
└── ai.py          ← 今回新しく追加

4. 準備

インストール

pip install anthropic

前回までのライブラリに anthropic を追加するだけです。

APIキーの設定

# .env ファイルに追記
SECRET_KEY=your-jwt-secret-key
ANTHROPIC_API_KEY=sk-ant-api03-...  ← Anthropicのコンソールで取得

⚠️ APIキーは絶対にコードに直書きしない
.env に書いて、.gitignore に .env を追加しておきましょう。
GitHubに上げると即座に不正利用されます。

5. 実装

Step 1：使用ログをDBに記録するためにmodels.pyを更新

誰がいつ何を送ったかを記録するテーブルを追加します。

# models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"
    id              = Column(Integer, primary_key=True, index=True)
    name            = Column(String, nullable=False)
    email           = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active       = Column(Boolean, default=True)

class AIUsageLog(Base):                          # ← 今回追加
    __tablename__ = "ai_usage_logs"

    id           = Column(Integer, primary_key=True, index=True)
    user_id      = Column(Integer, ForeignKey("users.id"), nullable=False)
    prompt       = Column(String, nullable=False)   # 送ったメッセージ
    response     = Column(String, nullable=False)   # AIの返答
    input_tokens = Column(Integer, default=0)       # 使ったトークン数（入力）
    output_tokens= Column(Integer, default=0)       # 使ったトークン数（出力）
    created_at   = Column(DateTime, server_default=func.now())

💡 なぜトークン数を記録するのか？
Claude APIはトークン（文字のかたまり）の量で課金されます。
誰がどれだけ使っているか把握するために記録しておくと、コスト管理に役立ちます。

Step 2：schemas.py にAI用の型を追加

# schemas.py（既存のものに追記）
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

# ── 既存のスキーマ（省略）──

# ── 今回追加 ─────────────────────────────────
class ChatRequest(BaseModel):
    message: str = Field(
        min_length=1,
        max_length=4000,
        description="AIに送るメッセージ"
    )
    system_prompt: Optional[str] = Field(
        default=None,
        max_length=1000,
        description="AIへの指示（省略可）"
    )

class ChatResponse(BaseModel):
    reply:         str
    input_tokens:  int
    output_tokens: int

class UsageLog(BaseModel):
    id:            int
    prompt:        str
    response:      str
    input_tokens:  int
    output_tokens: int
    created_at:    datetime
    model_config = {"from_attributes": True}

Step 3：ai.py ── Claude APIを呼ぶ処理を書く

これがこの記事の核心部分です。3つの関数を作ります。

# ai.py
import os
import anthropic
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from typing import List

from database import get_db
from auth import get_current_user
from models import AIUsageLog, User
import schemas

router = APIRouter(prefix="/ai", tags=["AI"])

# Anthropicクライアントの初期化
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL  = "claude-sonnet-4-6"

DEFAULT_SYSTEM = "あなたは親切なアシスタントです。日本語で回答してください。"


# ── 通常のチャット ─────────────────────────────────────────────────
@router.post("/chat", response_model=schemas.ChatResponse)
def chat(
    request: schemas.ChatRequest,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user)  # ログイン必須
):
    """メッセージを送ってAIの返答を受け取る"""
    system = request.system_prompt or DEFAULT_SYSTEM

    # Claude APIを呼ぶ
    response = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": request.message}]
    )

    reply         = response.content[0].text
    input_tokens  = response.usage.input_tokens
    output_tokens = response.usage.output_tokens

    # 使用ログをDBに保存
    log = AIUsageLog(
        user_id=current_user.id,
        prompt=request.message,
        response=reply,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
    )
    db.add(log)
    db.commit()

    return schemas.ChatResponse(
        reply=reply,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
    )


# ── ストリーミングチャット ──────────────────────────────────────────
@router.post("/chat/stream")
def chat_stream(
    request: schemas.ChatRequest,
    current_user: User = Depends(get_current_user)  # ログイン必須
):
    """AIの返答をリアルタイムで少しずつ受け取る（ストリーミング）"""
    system = request.system_prompt or DEFAULT_SYSTEM

    def generate():
        with client.messages.stream(
            model=MODEL,
            max_tokens=1024,
            system=system,
            messages=[{"role": "user", "content": request.message}]
        ) as stream:
            for text in stream.text_stream:
                yield text   # テキストが来るたびに少しずつ送る

    return StreamingResponse(generate(), media_type="text/plain")


# ── 使用状況の確認 ─────────────────────────────────────────────────
@router.get("/usage/me", response_model=List[schemas.UsageLog])
def get_my_usage(
    limit:        int     = 20,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user)
):
    """自分の使用履歴を取得する"""
    logs = (
        db.query(AIUsageLog)
        .filter(AIUsageLog.user_id == current_user.id)
        .order_by(AIUsageLog.created_at.desc())
        .limit(limit)
        .all()
    )
    return logs

💡 APIRouter とは？
エンドポイントをファイルごとに分けるための仕組みです。
prefix="/ai" を設定すると、このファイル内の全エンドポイントのURLが /ai/... から始まります。
main.py を肥大化させずに機能を整理できます。

Step 4：main.py にルーターを登録する

# main.py（抜粋）
from fastapi import FastAPI, Depends
from database import engine
import models
import ai   # ← 追加

models.Base.metadata.create_all(bind=engine)

app = FastAPI(title="認証付きAI API")

# ルーターを登録（この1行だけでai.pyのエンドポイントが全部有効になる）
app.include_router(ai.router)

# 既存のエンドポイント（省略）

6. ストリーミングとは？

通常のチャット（POST /ai/chat）は、AIが返答を全部作り終えてから一度に送ります。

ユーザー → リクエスト → AIが考える（数秒）→ 一気に返す

ストリーミング（POST /ai/chat/stream）は、返答が少しできるたびに送り続けます。

ユーザー → リクエスト → 「こんに」→「ちは」→「！今日は」→「…」と少しずつ届く

ChatGPTやClaudeの画面で文字が1文字ずつ表示されているのがストリーミングです。
待ち時間のストレスを下げる効果があります。

7. 動作確認

uvicorn main:app --reload

http://localhost:8000/docs で全エンドポイントを確認できます。

確認の手順

① POST /auth/login でログインしてトークンを取得
② Swagger UIの [Authorize] にトークンを貼り付ける

③ POST /ai/chat でメッセージを送る
   {"message": "FastAPIを一言で教えて"}
   → {"reply": "FastAPIはPythonで...", "input_tokens": 20, "output_tokens": 45}

④ GET /ai/usage/me で使用履歴を確認
   → [{id:1, prompt:"FastAPIを...", ...}]

⑤ トークンなしで POST /ai/chat を叩く
   → 401 Unauthorized ✅（ログイン必須が効いている）

8. よくある間違いと対処法

間違い1：APIキーを環境変数から読み込んでいない

# ✖ コードに直書き → GitHubに上げたら即アウト
client = anthropic.Anthropic(api_key="sk-ant-api03-xxxx")

# 〇 環境変数から読む
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

間違い2：`anthropic` ライブラリをインストール忘れ

ModuleNotFoundError: No module named 'anthropic'

pip install anthropic

間違い3：ストリーミングで `media_type` を間違える

# ✖ JSONで返そうとするとエラーになる
return StreamingResponse(generate(), media_type="application/json")

# 〇 テキストで返す
return StreamingResponse(generate(), media_type="text/plain")

間違い4：DBのテーブルが更新されない

AIUsageLog テーブルを追加した後、既存の app.db を削除して再起動が必要です。

rm app.db
uvicorn main:app --reload

9. 発展：このままポートフォリオに使える

今回作ったものをGitHubに上げるときのREADMEの書き方例です。

## 認証付きAI API

FastAPI + SQLite + Claude API を使って作った社内向けAIツールの土台。

### 主な機能
- JWT認証（登録・ログイン）
- Claude APIとのチャット
- ストリーミング対応
- 使用ログのDB保存（トークン数・ユーザー別）

### 使用技術
- FastAPI / SQLAlchemy / SQLite
- python-jose（JWT）/ passlib（bcrypt）
- anthropic（Claude API）

このリポジトリが「FastAPIの基礎 + 認証 + AI連携」の3軸を1つで示せる成果物になります。

10. シリーズを振り返って

6回にわたる学習日記の流れを整理します。

回	テーマ	身についたこと
#1	FastAPI概要	なぜFastAPIか・クイックスタート
#2	Pydantic	型バリデーション・Fieldの制約
#3	パラメータ	パス・クエリ・ボディの使い分け
#4	CRUD + DB	SQLAlchemy / SQLModel・DB操作
#5	JWT認証	パスワードハッシュ・トークン認証
#6	AI連携	外部API統合・ストリーミング・ログ

この6本で「設計 → 実装 → 認証 → AI統合」という実務に近い一連の流れを経験できました。

参考

この記事は学習日記として書いています。間違いや補足があればコメントいただけると嬉しいです！