認証・権限管理：本番想定の設計

本レッスンでは、MCP サーバーを マルチユーザー本番環境 で運用するための認証・認可を扱います。

MCP サーバーの典型的な認証ニーズ

認証フローの設計

# MCP サーバー起動時に資格情報を渡すパターン

# Claude Desktop config:
{
  "mcpServers": {
    "myapp": {
      "command": "python",
      "args": ["server.py"],
      "env": {
        "API_KEY": "sk-...",
        "USER_ID": "user_12345"
      }
    }
  }
}

# サーバー側で取得
import os
API_KEY = os.environ["API_KEY"]
USER_ID = os.environ["USER_ID"]

OAuth フローの実装

MCP サーバーが SaaS（Slack/Notion/Google）を呼ぶ場合、ユーザーごとの OAuth トークンが必要です。

# oauth_mcp.py
from mcp.server.fastmcp import FastMCP
import requests

mcp = FastMCP("oauth-mcp")

# 起動時に必要な情報
USER_ID = os.environ["USER_ID"]
TOKEN_STORE_URL = os.environ["TOKEN_STORE_URL"]


def get_user_token(user_id: str, provider: str) -> str:
    """セキュアなトークン保管庫から取得。"""
    response = requests.get(
        f"{TOKEN_STORE_URL}/tokens",
        params={"user_id": user_id, "provider": provider},
        headers={"X-Service-Auth": os.environ["SERVICE_AUTH_KEY"]},
    )
    response.raise_for_status()
    return response.json()["access_token"]


@mcp.tool()
def post_to_slack(channel: str, message: str) -> str:
    """ユーザーのSlackに投稿（OAuth認証）。"""
    token = get_user_token(USER_ID, "slack")
    response = requests.post(
        "https://slack.com/api/chat.postMessage",
        headers={"Authorization": f"Bearer {token}"},
        json={"channel": channel, "text": message},
    )
    if response.json().get("ok"):
        return "投稿完了"
    return f"エラー: {response.json().get('error')}"

権限スコープの管理

# スコープ定義
SCOPES = {
    "read_only": ["search", "list", "get"],
    "write": ["create", "update"],
    "admin": ["delete", "manage_users"],
}


def require_scope(scope: str):
    """ツールに必要なスコープをデコレータで強制。"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_scopes = get_user_scopes(USER_ID)
            if scope not in user_scopes:
                return f"権限不足: {scope} が必要です"
            return func(*args, **kwargs)
        return wrapper
    return decorator


@mcp.tool()
@require_scope("write")
def create_record(data: dict) -> str:
    """レコード作成（write 権限が必要）。"""
    ...


@mcp.tool()
@require_scope("admin")
def delete_user(user_id: int) -> str:
    """ユーザー削除（admin 権限が必要）。"""
    ...

マルチテナント分離

# テナント分離の実装

@mcp.tool()
def search_records(query: str) -> str:
    """レコード検索（テナント内のみ）。"""
    tenant_id = get_user_tenant(USER_ID)

    # 必ずテナントIDで絞り込み
    sql = """
        SELECT * FROM records
        WHERE tenant_id = %s AND query MATCH %s
        LIMIT 100
    """
    return execute_query(sql, [tenant_id, query])


# DB レベルでも RLS（Row Level Security）を設定
"""
ALTER TABLE records ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON records
    USING (tenant_id = current_setting('app.current_tenant_id')::int);

# クエリ前にテナントを設定
SET app.current_tenant_id = '12345';
"""

PII（個人情報）保護

# PII を含む結果を自動マスキング

import re

def mask_pii(text: str) -> str:
    # メールアドレス
    text = re.sub(r'b[w.-]+@[w.-]+.w+b', '[EMAIL]', text)
    # 電話番号（日本）
    text = re.sub(r'b0d{1,4}-d{1,4}-d{3,4}b', '[PHONE]', text)
    # クレジットカード
    text = re.sub(r'bd{4}[-s]?d{4}[-s]?d{4}[-s]?d{4}b', '[CC]', text)
    # マイナンバー（12桁）
    text = re.sub(r'bd{12}b', '[MY_NUMBER]', text)
    return text


@mcp.tool()
def search_customer_logs(query: str) -> str:
    raw_result = _do_search(query)
    return mask_pii(raw_result)

監査ログの実装

# 全ツール呼び出しを監査ログに記録

def audit_log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "user_id": USER_ID,
            "tool": func.__name__,
            "args": redact_sensitive(kwargs),
        }
        try:
            result = func(*args, **kwargs)
            log_entry["status"] = "success"
            log_entry["result_size"] = len(str(result))
        except Exception as e:
            log_entry["status"] = "error"
            log_entry["error"] = str(e)
            raise
        finally:
            audit_db.insert(log_entry)

        return result
    return wrapper


@mcp.tool()
@audit_log
def sensitive_operation(args):
    ...

セキュリティチェックリスト

失敗パターン

このレッスンのまとめ

本番運用のMCPサーバーは「OAuth → スコープ → テナント分離 → PII保護 → 監査」が必須。次のレッスンでは、Claude Desktop / API クライアント側の実装を学びます。