LoginSignup
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

SupabaseでリアルタイムAIを構築する | 第5章：最適化とコミュニティ

Posted at

はじめに

これまでの第1章から第4章では、SupabaseとModel Context Protocol（MCP）を活用してリアルタイムAIを構築する方法を学びました。プロジェクトデータの取得、リアルタイムサブスクリプション、データ分析エージェント、そしてコミュニティ向けチャットボットまで、Supabaseの強力なデータベースとMCPの柔軟性を組み合わせたエージェントAIを実装しました。この最終章では、MCPサーバーの最適化、セキュリティ強化、そしてコミュニティへの貢献に焦点を当てます。

この第5章では、Supabase用MCPサーバーのパフォーマンスを向上させ、安全性を確保する方法を解説します。さらに、サーバーをオープンソースとして共有し、MCPコミュニティに貢献する方法を探ります。コード例を通じて、セキュリティログの実装も学びます。SupabaseとMCPの未来を一緒に切り開きましょう！

MCPサーバーの最適化

Supabase用MCPサーバーを本番環境で運用するには、以下の最適化が必要です：

1. キャッシュの活用

  • 目的：頻繁なデータ取得（例：プロジェクト統計）を高速化。
  • 方法：Redisなどのインメモリキャッシュを使用。
  • 例：プロジェクト統計を5分間キャッシュし、Supabaseへのリクエストを削減。

2. レートリミティング

  • 目的：APIの過剰使用やDoS攻撃を防止。
  • 方法：リクエストごとに制限を設定（例：1分間に100リクエスト）。
  • 例：ratelimitライブラリを使用して制限を実装。

3. 非同期処理

  • 目的：大量のリクエストやリアルタイムイベントを効率的に処理。
  • 方法：asyncioやメッセージキュー（例：RabbitMQ）を活用。
  • 例：チャットボットのメッセージ処理をキューイング。

セキュリティ強化

リアルタイムAIを安全に運用するには、以下のセキュリティ対策が重要です：

1. HTTPSの有効化

  • 目的：通信データを暗号化し、盗聴を防止。
  • 方法：Let’s EncryptやクラウドプロバイダーのSSL証明書を使用。
  • 推奨事項：本番環境ではTLS 1.3を採用。

2. OAuth 2.1認証

  • 目的：SupabaseやMCPサーバーへのアクセスを制限。
  • 方法：SupabaseのJWTトークンを使用し、auth: noneを避ける。
  • 例：ユーザーの認証情報を確認後、MCPリクエストを処理。

3. セキュリティログ

  • 目的：不正アクセスやエラーを追跡。
  • 方法：リクエストとレスポンスをログに記録。

コード例：最適化とセキュリティログの実装

以下のコードは、第4章のチャットボットサーバーにキャッシュ（Redis）とセキュリティログを追加した例です：

from mcp import MCPServer
from supabase import create_client
import os
from dotenv import load_dotenv
import redis
import logging
from datetime import datetime
import json

# ログ設定
logging.basicConfig(
    filename="chatbot_server.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

class OptimizedChatbotServer(MCPServer):
    def __init__(self, host, port, supabase_url, supabase_key, redis_host, redis_port):
        super().__init__(host, port)
        self.supabase = create_client(supabase_url, supabase_key)
        self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
        self.logger = logging.getLogger(__name__)
        self.register_resource("get_recent_messages", self.get_recent_messages)
        self.register_tool("send_message", self.send_message)

    def get_recent_messages(self, params):
        request_id = datetime.now().isoformat()
        self.logger.info(f"リクエスト受信 [ID: {request_id}]: パラメータ={params}")
        try:
            limit = params.get("limit", 10)
            project_id = params.get("project_id", None)
            cache_key = f"messages:{project_id}:{limit}"
            
            # キャッシュを確認
            cached = self.redis.get(cache_key)
            if cached:
                self.logger.info(f"キャッシュヒット [ID: {request_id}]: キー={cache_key}")
                return {"status": "success", "messages": json.loads(cached)}
            
            # Supabaseから取得
            query = self.supabase.table("chat_messages").select("*").order("created_at", desc=True).limit(limit)
            if project_id:
                query = query.eq("project_id", project_id)
            response = query.execute()
            messages = response.data
            
            # キャッシュに保存（5分）
            self.redis.setex(cache_key, 300, json.dumps(messages))
            self.logger.info(f"リクエスト成功 [ID: {request_id}]: レスポンス={len(messages)}件")
            return {"status": "success", "messages": messages}
        except Exception as e:
            self.logger.error(f"リクエスト失敗 [ID: {request_id}]: エラー={str(e)}")
            return {"status": "error", "message": str(e)}

    def send_message(self, params):
        request_id = datetime.now().isoformat()
        self.logger.info(f"リクエスト受信 [ID: {request_id}]: パラメータ={params}")
        try:
            user_id = params.get("user_id", "ai_bot")
            message = params.get("message", "")
            project_id = params.get("project_id", None)
            if not message:
                self.logger.warning(f"リクエスト失敗 [ID: {request_id}]: メッセージが空")
                return {"status": "error", "message": "メッセージが必要です"}
            data = {"user_id": user_id, "message": message}
            if project_id:
                data["project_id"] = project_id
            response = self.supabase.table("chat_messages").insert(data).execute()
            self.logger.info(f"リクエスト成功 [ID: {request_id}]: メッセージID={response.data[0]['id']}")
            return {"status": "success", "message_id": response.data[0]["id"]}
        except Exception as e:
            self.logger.error(f"リクエスト失敗 [ID: {request_id}]: エラー={str(e)}")
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = OptimizedChatbotServer(
        host="localhost",
        port=8098,
        supabase_url=os.getenv("SUPABASE_URL"),
        supabase_key=os.getenv("SUPABASE_KEY"),
        redis_host=os.getenv("REDIS_HOST", "localhost"),
        redis_port=int(os.getenv("REDIS_PORT", 6379))
    )
    print("最適化チャットボットMCPサーバーを起動中: http://localhost:8098")
    server.start()

コードの説明

  • Redisキャッシュ：メッセージ取得をキャッシュ（5分間有効）。setexで有効期限を設定。
  • セキュリティログ：リクエストの受信、成功、失敗をchatbot_server.logに記録。
  • エラーハンドリング：空のメッセージや例外を適切に処理。
  • 環境変数：Redisのホストとポートを追加（.envにREDIS_HOST、REDIS_PORTを設定）。

前提条件

  • Redisサーバーがローカルまたはクラウドで稼働（例：docker run -p 6379:6379 redis）。
  • .envファイルにSUPABASE_URL、SUPABASE_KEY、およびREDIS_HOST、REDIS_PORTが設定済み。
  • chat_messagesテーブルがSupabaseに存在。

テスト方法

  1. Redis起動： 
    docker run -p 6379:6379 redis
  2. サーバー起動： 
    python optimized_chatbot_server.py
  3. メッセージ取得のテスト： 
    import requests
import json

url = "http://localhost:8098"
payload = {
    "jsonrpc": "2.0",
    "method": "get_recent_messages",
    "params": {"limit": 5, "project_id": 1},
    "id": 1
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))
  4. ログ確認：
    chatbot_server.logに以下のような記録が残る： 
    2025-04-16 23:30:00,123 - INFO - リクエスト受信 [ID: 2025-04-16T23:30:00.123456]: パラメータ={'limit': 5, 'project_id': 1}
2025-04-16 23:30:00,125 - INFO - リクエスト成功 [ID: 2025-04-16T23:30:00.123456]: レスポンス=2件

コミュニティへの貢献

Supabase用MCPサーバーをオープンソースとして共有することで、コミュニティに貢献できます。以下のステップで進めます：

1. GitHubでの公開

  • リポジトリ作成：GitHubに新しいリポジトリを作成（例：supabase-mcp-server）。

  • コード整理：モジュール化し、再利用可能な構造にする。

  • README：インストール手順、使い方、例を記載。

    # Supabase MCP Server
SupabaseとMCPを統合し、リアルタイムAIを構築するサーバーです。

## インストール
```bash
pip install mcp supabase redis

    使い方

    1. .envにSUPABASE_URLとSUPABASE_KEYを設定。
    2. python server.pyで起動。

  • ライセンス：MITやApache 2.0など、オープンソースライセンスを選択。

2. ドキュメントの提供

  • Qiita記事：このシリーズのようなチュートリアルを共有。
  • Supabaseコミュニティ：SupabaseのDiscordやフォーラムでプロジェクトを紹介。
  • MCPコミュニティ：MCPのGitHub IssuesやDiscordでサーバーを提案。

3. フィードバックの収集

  • Issueトラッキング：バグ報告や機能リクエストを受け付ける。
  • プルリクエスト：他の開発者からの貢献を歓迎。
  • 改善の継続：コミュニティのフィードバックを基にサーバーを更新。

SupabaseとMCPの未来

SupabaseとMCPの組み合わせは、リアルタイムAIの可能性を大きく広げます。以下は、長期的なビジョンです：

1. シームレスな統合

  • ビジョン：SupabaseがMCPをネイティブサポートし、ダッシュボードからMCPサーバーを設定可能に。
  • 例：SupabaseのAPIエンドポイントが直接MCPリソースとして登録可能。

2. エンタープライズ採用

  • ビジョン：企業がSupabaseとMCPを使って、カスタマーサポートやデータ分析を自動化。
  • 例：CRMデータをSupabaseに保存し、AIが顧客対応をリアルタイムで最適化。

3. 個人向けAIアシスタント

  • ビジョン：個人がSupabaseにプライベートデータを保存し、MCP経由でパーソナライズされたAIを利用。
  • 例：個人のタスク管理や学習ログをAIが分析し、スケジュールを提案。

シリーズのまとめ

このシリーズを通じて、SupabaseとMCPを活用したリアルタイムAIの構築を以下のように学びました：

  • 第1章：SupabaseとMCPの基本、プロジェクトデータ取得。
  • 第2章：リアルタイムサブスクリプションでデータ変更を捕捉。
  • 第3章：データ分析エージェントで進捗や傾向を分析。
  • 第4章：チャットボットでコミュニティ対話を強化。
  • 第5章：サーバーの最適化、セキュリティ、コミュニティ貢献。

SupabaseのリアルタイムデータベースとMCPの標準化された接続は、AIを動的かつ実用的なアシスタントに変える力を持っています。あなたもこのサーバーを試し、コミュニティで共有して、リアルタイムAIの未来を共創しませんか？

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の挑戦でまたお会いしましょう！

2
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?