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を構築する | 第4章：コミュニティAIの強化：チャットボットの構築

Posted at

はじめに

第3章では、Supabaseのデータを活用してデータ分析エージェントを構築し、プロジェクトの進捗や傾向を分析するMCPサーバーを実装しました。これにより、AIが統計を生成し、優先度の提案を行うことができました。今回は、SupabaseとModel Context Protocol（MCP）を使って、コミュニティを強化するリアルタイムチャットボットを構築します。このチャットボットは、Supabaseに保存されたチャットログを基に、ユーザーからの質問に即座に回答します。

この第4章では、Supabaseにチャットメッセージを保存し、MCPサーバーを通じてAIがメッセージを取得・処理する仕組みを作ります。たとえば、ユーザーが「プロジェクトAの状況は？」と尋ねると、AIがSupabaseのデータから関連情報を抽出し、適切な回答を返します。コード例とステップごとのガイドで、コミュニティAIの構築を体験しましょう。さあ、始めましょう！

コミュニティAIとチャットボットの概要

コミュニティAIは、ユーザー間の対話を支援し、情報を提供するエージェントです。Supabaseをデータストアとして使用することで、チャットログを効率的に管理し、リアルタイムでAIに提供できます。MCPサーバーは、以下の役割を果たします：

  • メッセージ保存：ユーザーのメッセージをSupabaseに記録。
  • メッセージ取得：最新のチャットログをAIに提供。
  • アクション実行：AIが生成した回答をSupabaseに保存または外部ツール（例：Slack）に送信。

ユースケース

  • コミュニティ管理：DiscordやSlackでユーザーからの質問に自動回答。
  • カスタマーサポート：問い合わせをリアルタイムで処理し、履歴を保存。
  • チームコラボレーション：プロジェクト関連の質問に即座に答える。

開発環境の準備

第3章の環境を基に、以下の準備を行います：

  • Python 3.8以降、mcpライブラリ、supabaseライブラリ、Claude Desktop：これまでと同じ。
  • Supabaseテーブル：チャットメッセージを保存する新しいテーブルを作成。
  • python-dotenv：環境変数の管理（既にインストール済み）。

Supabaseテーブルの作成

チャットログを保存するchat_messagesテーブルを作成します。SQLエディタで以下を実行：

CREATE TABLE chat_messages (
  id SERIAL PRIMARY KEY,
  user_id TEXT NOT NULL,
  message TEXT NOT NULL,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  project_id INTEGER REFERENCES projects(id)
);

テストデータを挿入：

INSERT INTO chat_messages (user_id, message, project_id) VALUES
('user1', 'プロジェクトAの進捗はどう？', 1),
('user2', 'タスクの期限を確認したい', 1),
('user3', 'プロジェクトBのデータ分析について', 2);

環境変数

第1章の.envファイル（SUPABASE_URL、SUPABASE_KEY）を再利用。

コード例：チャットボット用MCPサーバー

以下のMCPサーバーは、Supabaseのchat_messagesテーブルにメッセージを保存し、最新のメッセージを取得する機能を提供します。さらに、プロジェクトデータと連携してコンテキストを生成します。

from mcp import MCPServer
from supabase import create_client
import os
from dotenv import load_dotenv

class ChatbotServer(MCPServer):
    def __init__(self, host, port, supabase_url, supabase_key):
        super().__init__(host, port)
        self.supabase = create_client(supabase_url, supabase_key)
        self.register_resource("get_recent_messages", self.get_recent_messages)
        self.register_tool("send_message", self.send_message)

    def get_recent_messages(self, params):
        try:
            limit = params.get("limit", 10)
            project_id = params.get("project_id", None)
            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
            return {"status": "success", "messages": messages}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def send_message(self, params):
        try:
            user_id = params.get("user_id", "ai_bot")
            message = params.get("message", "")
            project_id = params.get("project_id", None)
            if not message:
                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()
            return {"status": "success", "message_id": response.data[0]["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = ChatbotServer(
        host="localhost",
        port=8097,
        supabase_url=os.getenv("SUPABASE_URL"),
        supabase_key=os.getenv("SUPABASE_KEY")
    )
    print("チャットボットMCPサーバーを起動中: http://localhost:8097")
    server.start()

コードの説明

  • get_recent_messages：最新のチャットメッセージを取得。project_idでフィルタリング可能。
  • send_message：AIまたはユーザーが新しいメッセージをSupabaseに保存。
  • register_resource/tool：メッセージ取得をリソース、送信をツールとして登録。
  • データ構造：chat_messagesテーブルにユーザーID、メッセージ、プロジェクトIDを保存。

前提条件

  • chat_messagesとprojectsテーブルがSupabaseに存在。
  • .envファイルに正しいSUPABASE_URLとSUPABASE_KEYが設定済み。

サーバーのテスト

サーバーが正しく動作するか確認します：

  1. サーバー起動：

    python chatbot_server.py

    コンソールに「チャットボットMCPサーバーを起動中: http://localhost:8097」と表示。

  2. メッセージ取得のテスト：
    Pythonでリクエストを送信：

    import requests
import json

url = "http://localhost:8097"
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))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "messages": [
      {
        "id": 1,
        "user_id": "user1",
        "message": "プロジェクトAの進捗はどう？",
        "created_at": "2025-04-16T00:00:00",
        "project_id": 1
      },
      {
        "id": 2,
        "user_id": "user2",
        "message": "タスクの期限を確認したい",
        "created_at": "2025-04-16T00:00:00",
        "project_id": 1
      }
    ]
  },
  "id": 1
}

  3. メッセージ送信のテスト：

    payload = {
    "jsonrpc": "2.0",
    "method": "send_message",
    "params": {
        "user_id": "ai_bot",
        "message": "プロジェクトAの進捗を確認中です",
        "project_id": 1
    },
    "id": 2
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "message_id": 4
  },
  "id": 2
}

Claude Desktopとの接続

サーバーをClaude Desktopに接続します：

  1. 設定ファイルの編集：
    Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：

    {
  "mcp_servers": [
    {
      "name": "ChatbotServer",
      "url": "http://localhost:8097",
      "auth": "none"
    }
  ]
}

  2. Claudeでテスト：
    Claude Desktopを起動し、プロンプトを入力：

    プロジェクトAの最近のメッセージを教えてください。

    レスポンス例：

    プロジェクトAの最近のメッセージ：
- ユーザー1：プロジェクトAの進捗はどう？
- ユーザー2：タスクの期限を確認したい

    別のプロンプト：

    プロジェクトAについて「進捗を確認しました」と返信してください。

    レスポンス例：

    メッセージ「進捗を確認しました」をプロジェクトAに送信しました。

エージェントロジックの構築

実際のチャットボットでは、メッセージを処理して適切な回答を生成するロジックが必要です。以下は、簡略化したロジック例：

def process_message(self, message, project_id):
    if "進捗" in message:
        stats = self.supabase.table("tasks").select("status").eq("project_id", project_id).execute().data
        completed = len([t for t in stats if t["status"] == "completed"])
        total = len(stats)
        response = f"プロジェクトの進捗：{completed}/{total} タスク完了"
        self.send_message({"user_id": "ai_bot", "message": response, "project_id": project_id})
        return response
    return "処理可能な質問をしてください"

このロジックは、「進捗」というキーワードを含むメッセージに対して、タスクの完了状況を返します。

実装のコツと注意点

  • メッセージフィルタリング：スパムや不適切なメッセージを防ぐフィルタを追加。
  • リアルタイム統合：第2章のサブスクリプションを活用し、新着メッセージを即座に処理。
  • セキュリティ：本番環境では、ユーザー認証を追加し、auth: noneを避ける。
  • テスト：テスト用テーブルで動作確認し、本番データに影響を与えない。
  • スケーラビリティ：大量のメッセージを処理する場合、キャッシュやキュー（例：Redis）を検討。

試してみよう：挑戦課題

以下の機能を追加して、チャットボットを強化してみてください：

  • メッセージに「タスク追加」とある場合、自動でtasksテーブルにタスクを追加。
  • 特定のキーワード（例：ヘルプ）を検出し、定型回答を返す機能。
  • メッセージを外部ツール（例：Slack）に転送するツールを登録。

まとめと次のステップ

この第4章では、Supabaseを活用してリアルタイムチャットボットを構築しました。チャットログの保存と取得を通じて、AIがコミュニティの対話に参加できるようになりました。

次の第5章では、MCPサーバーの最適化とコミュニティへの貢献に焦点を当てます。サーバーのパフォーマンス向上、セキュリティ強化、そしてSupabase用MCPサーバーをオープンソースとして共有する方法を学びます。コミュニティ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?