Python・FastAPIでAIボット検知ミドルウェアを実装する

FastAPIでAIボット対策する意義

PythonのAPIバックエンドやWebアプリにもAIクローラー対策は必要です。FastAPIのMiddlewareを使えば、すべてのルートに一括で適用できます。

基本実装：全AIボットをHTTP 402で拒否

from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import re

AI_BOT_PATTERN = re.compile(
    r"GPTBot|OAI-SearchBot|ChatGPT-User|"
    r"ClaudeBot|anthropic-ai|PerplexityBot|"
    r"Google-Extended|Bytespider|CCBot|"
    r"cohere-ai|YouBot|Diffbot|Amazonbot|MistralBot",
    re.IGNORECASE
)

class AIBotBlockerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ua = request.headers.get("user-agent", "")
        if AI_BOT_PATTERN.search(ua):
            return Response(
                content='{"error":"License required","url":"https://monitor.microforge.works/pricing"}',
                status_code=402,
                media_type="application/json"
            )
        return await call_next(request)

app = FastAPI()
app.add_middleware(AIBotBlockerMiddleware)

ライセンスキー検証付きTollgate実装

from cachetools import TTLCache
import httpx

license_cache: TTLCache = TTLCache(maxsize=1000, ttl=300)

async def verify_license(api_key: str, site_id: str) -> bool:
    cache_key = f"{site_id}:{api_key}"
    if cache_key in license_cache:
        return license_cache[cache_key]
    async with httpx.AsyncClient() as client:
        r = await client.get(
            "https://monitor-api.microforge.works/license/verify",
            params={"site_id": site_id, "api_key": api_key},
            timeout=2.0
        )
        valid = r.status_code == 200 and r.json().get("valid")
    license_cache[cache_key] = valid
    return valid

class TollgateMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ua = request.headers.get("user-agent", "")
        if not AI_BOT_PATTERN.search(ua):
            return await call_next(request)

        api_key = request.headers.get("x-license-key", "")
        site_id = "YOUR_SITE_ID"

        if api_key and await verify_license(api_key, site_id):
            return await call_next(request)

        return Response(
            content='{"error":"Payment required"}',
            status_code=402,
            headers={"X-License-URL": "https://monitor.microforge.works/pricing"}
        )

ログ記録との組み合わせ

import logging

logger = logging.getLogger(__name__)

class AIBotLoggerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ua = request.headers.get("user-agent", "")
        if AI_BOT_PATTERN.search(ua):
            logger.info(
                "AI bot blocked: ua=%s ip=%s path=%s",
                ua,
                request.client.host,
                request.url.path
            )
        return await call_next(request)

AI Access Monitorとの連携

FastAPIバックエンドのアクセスも含めて一元管理したい場合、AI Access MonitorのAPI（POST /collect）を呼び出すことで、ダッシュボードに統合できます。