はじめに
前回までで、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統合」という実務に近い一連の流れを経験できました。
参考
この記事は学習日記として書いています。間違いや補足があればコメントいただけると嬉しいです！