FastAPI で JWT + Cognito 認証を実装する

2026年9月27日

FastAPI JWT 認証 – AWS Cognito との連携実装ガイド

JWT と Cognito について

セッション認証（従来の username/password + session cookie）から JWT（JSON Web Token）認証へ移行しつつある時代。さらに AWS を使う場合は Cognito（マネージド認証サービス）と組み合わせることで、セキュリティと保守性が向上します。

今回は、FastAPI + JWT + Cognito を組み合わせた認証実装についてまとめておきたいと思います。

もしご自身で先に公式ドキュメントをご覧になられる方は下のリンクから眺めてみてください。
AWS Cognito Documentation ー公式ドキュメント（英語）

対象となる方

FastAPI で本格的な認証を実装したい方
AWS Cognito を使った認証に興味のある方
セキュリティを意識した実装を学びたい方

　※ このドキュメントは FastAPI 0.100 以上、AWS Cognito を使用します。

3 つの認証方式の比較

セッション認証（従来）

@app.post("/login")
async def login(email: str, password: str):
    user = await db.query(User).filter(User.email == email).first()
    if not user or not verify_password(password, user.password_hash):
        raise HTTPException(status_code=401)

    session_id = str(uuid4())
    sessions[session_id] = {"user_id": user.id}
    response.set_cookie("session_id", session_id)
    return {"success": True}

問題： サーバーがセッションを保持する必要があるため、スケーリングが困難。

JWT 認証（ステートレス）

from datetime import datetime, timedelta
import jwt

@app.post("/login")
async def login(email: str, password: str):
    user = await db.query(User).filter(User.email == email).first()
    if not user or not verify_password(password, user.password_hash):
        raise HTTPException(status_code=401)

    payload = {
        "sub": user.id,
        "email": user.email,
        "exp": datetime.utcnow() + timedelta(hours=1),
    }
    token = jwt.encode(payload, "secret_key", algorithm="HS256")
    return {"access_token": token}

利点： サーバーは状態を保持しない（ステートレス）。スケーリングが容易。

Cognito（マネージド）

import httpx
import jwt

@app.post("/login")
async def login(email: str, password: str):
    cognito_response = await authenticate_with_cognito(email, password)
    id_token = cognito_response["id_token"]
    return {"access_token": id_token}

@app.get("/api/me")
async def get_me(token: str = Header(...)):
    jwks = await fetch_cognito_jwks()
    payload = jwt.decode(
        token,
        jwks,
        algorithms=["RS256"],
        audience="cognito_client_id"
    )
    return {"user_id": payload.get("sub")}

利点： ユーザー管理が AWS に任せられる。MFA・ソーシャルログイン対応。

実装例：Cognito との連携

では、Cognito と連携した実装を見てみましょう。

import os
import httpx
import jwt
from fastapi import HTTPException, Header, Depends
from dataclasses import dataclass

@dataclass
class CurrentUser:
    user_id: str
    email: str
    sub: str

async def fetch_cognito_jwks():
    pool_id = os.environ["AWS_COGNITO_USER_POOL_ID"]
    region = pool_id.split("_")[0]
    url = f"https://cognito-idp.{region}.amazonaws.com/{pool_id}/.well-known/jwks.json"

    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

async def verify_cognito_token(token: str) -> dict:
    pool_id = os.environ["AWS_COGNITO_USER_POOL_ID"]
    client_id = os.environ["AWS_COGNITO_CLIENT_ID"]
    region = pool_id.split("_")[0]

    try:
        jwks = await fetch_cognito_jwks()
        payload = jwt.decode(
            token,
            jwks,
            algorithms=["RS256"],
            audience=client_id,
            issuer=f"https://cognito-idp.{region}.amazonaws.com/{pool_id}",
        )

        if payload.get("token_use") != "id":
            raise HTTPException(status_code=401, detail="Invalid token type")

        return payload

    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

async def get_current_user(
    authorization: str = Header(...),
) -> CurrentUser:
    try:
        scheme, token = authorization.split()
        if scheme.lower() != "bearer":
            raise HTTPException(status_code=401, detail="Invalid auth scheme")
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid authorization header")

    payload = await verify_cognito_token(token)
    email = payload.get("email")
    sub = payload.get("sub")

    return CurrentUser(user_id=sub, email=email, sub=sub)

from fastapi import FastAPI, Depends
from auth import get_current_user, CurrentUser

app = FastAPI()

@app.get("/api/me")
async def get_me(user: CurrentUser = Depends(get_current_user)):
    return {
        "user_id": user.user_id,
        "email": user.email,
    }

@app.get("/api/protected")
async def protected_endpoint(user: CurrentUser = Depends(get_current_user)):
    return {"message": f"Hello, {user.email}", "user_id": user.user_id}

Cognito 設定

FastAPI で Cognito を使う際の環境変数：

AWS_COGNITO_USER_POOL_ID=ap-northeast-1_xxxxxxxxxxxxx
AWS_COGNITO_CLIENT_ID=xxxxxxxxxxxxxxxxx

認証方式の比較表

方式	実装複雑度	スケーラビリティ	セキュリティ
セッション認証	⭐	低い	⭐⭐
JWT 認証	⭐⭐	高い	⭐⭐⭐
Cognito	⭐⭐⭐	高い	⭐⭐⭐⭐