はじめに
第2章では、Supabaseのリアルタイム機能（WebSocketサブスクリプション）を活用し、**Model Context Protocol（MCP）**サーバーを構築してデータ変更を即座にAIに通知する方法を学びました。これにより、プロジェクト追加などのイベントをリアルタイムで捕捉できました。今回は、この基盤をさらに進化させ、データ分析エージェントを構築します。このエージェントは、Supabaseのデータを分析し、プロジェクトの進捗や傾向に関するインサイトを生成します。
この第3章では、Supabaseのデータを基に統計を計算するMCPサーバーを実装し、AIがプロジェクトの状況を要約したり、優先度の高いタスクを提案したりできるようにします。たとえば、「どのプロジェクトが最も活発か？」や「近日中のタスクは？」といった質問にAIが答えることができます。コード例とステップごとのガイドで、データ分析AIの構築を体験しましょう。さあ、始めましょう！
データ分析エージェントとは？
データ分析エージェントは、データベースから情報を抽出し、統計や傾向を計算してユーザーにインサイトを提供するAIです。SupabaseとMCPを組み合わせることで、以下のような機能を実現できます：
- 進捗分析：プロジェクトごとのタスク数や完了率を計算。
- 傾向分析：どのプロジェクトが最近活発か、どのカテゴリが優先されているかを特定。
- 提案生成：データに基づいて次のアクションを提案（例：遅延リスクのあるタスクの優先）。
ユースケース
- プロジェクト管理：チームの生産性を可視化し、ボトルネックを特定。
- ビジネス分析：売上データや顧客データを分析し、戦略を提案。
- コミュニティ管理：ユーザーの活動パターンを分析し、エンゲージメントを向上。
開発環境の準備
第2章の環境を基に、以下の追加準備を行います：
- Python 3.8以降、mcpライブラリ、supabaseライブラリ、Claude Desktop：第1章・第2章と同じ。
- pandasライブラリ：データ分析用。
- Supabaseテーブル：分析用のデータを格納する新しいテーブルを追加。
インストールコマンド：
pip install pandas
Supabaseテーブルの拡張
第1章のprojectsテーブルに加え、タスクデータを管理するtasksテーブルを作成します。SQLエディタで以下を実行：
CREATE TABLE tasks (
id SERIAL PRIMARY KEY,
project_id INTEGER REFERENCES projects(id),
name TEXT NOT NULL,
status TEXT CHECK (status IN ('open', 'in_progress', 'completed')),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
due_date TIMESTAMP
);
テストデータを挿入：
INSERT INTO tasks (project_id, name, status, due_date) VALUES
(1, 'ドキュメント作成', 'in_progress', '2025-04-18T00:00:00'),
(1, 'コードレビュー', 'open', '2025-04-20T00:00:00'),
(2, 'データ分析', 'completed', '2025-04-15T00:00:00');
環境変数
第1章の.envファイル（SUPABASE_URL、SUPABASE_KEY）を再利用。
コード例：データ分析用MCPサーバー
以下のMCPサーバーは、Supabaseのprojectsとtasksテーブルからデータを取得し、プロジェクトごとの統計を計算します。たとえば、タスク数や完了率を提供します。
from mcp import MCPServer
from supabase import create_client
import os
from dotenv import load_dotenv
import pandas as pd
from datetime import datetime
class AnalysisServer(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_project_stats", self.get_project_stats)
def get_project_stats(self, params):
try:
# プロジェクトとタスクを取得
projects = self.supabase.table("projects").select("*").execute().data
tasks = self.supabase.table("tasks").select("*").execute().data
# pandasでデータ分析
df_tasks = pd.DataFrame(tasks)
df_projects = pd.DataFrame(projects)
# プロジェクトごとの統計
stats = []
for project in projects:
project_tasks = df_tasks[df_tasks["project_id"] == project["id"]]
total_tasks = len(project_tasks)
completed_tasks = len(project_tasks[project_tasks["status"] == "completed"])
completion_rate = (completed_tasks / total_tasks * 100) if total_tasks > 0 else 0
overdue_tasks = len(project_tasks[
(project_tasks["status"] != "completed") &
(project_tasks["due_date"] < datetime.utcnow().isoformat())
])
stats.append({
"project_id": project["id"],
"project_name": project["name"],
"total_tasks": total_tasks,
"completed_tasks": completed_tasks,
"completion_rate": round(completion_rate, 2),
"overdue_tasks": overdue_tasks
})
return {"status": "success", "stats": stats}
except Exception as e:
return {"status": "error", "message": str(e)}
if __name__ == "__main__":
load_dotenv()
server = AnalysisServer(
host="localhost",
port=8096,
supabase_url=os.getenv("SUPABASE_URL"),
supabase_key=os.getenv("SUPABASE_KEY")
)
print("分析MCPサーバーを起動中: http://localhost:8096")
server.start()
コードの説明
-
get_project_stats：
projectsとtasksテーブルからデータを取得し、pandasで分析。 -
統計項目：
- 総タスク数（
total_tasks） - 完了タスク数（
completed_tasks） - 完了率（
completion_rate） - 期限切れタスク数（
overdue_tasks）
- 総タスク数（
- pandas：データフレームを使用して、プロジェクトごとのフィルタリングと計算を効率化。
- register_resource：統計リソースを登録し、AIがアクセス可能に。
前提条件
-
projectsとtasksテーブルがSupabaseに存在。 -
.envファイルに正しいSUPABASE_URLとSUPABASE_KEYが設定済み。
サーバーのテスト
サーバーが正しく動作するか確認します：
-
サーバー起動：
python analysis_server.pyコンソールに「分析MCPサーバーを起動中: http://localhost:8096」と表示。
-
統計の取得：
Pythonでリクエストを送信：import requests import json url = "http://localhost:8096" payload = { "jsonrpc": "2.0", "method": "get_project_stats", "params": {}, "id": 1 } response = requests.post(url, json=payload) print(json.dumps(response.json(), indent=2, ensure_ascii=False))期待されるレスポンス：
{ "jsonrpc": "2.0", "result": { "status": "success", "stats": [ { "project_id": 1, "project_name": "プロジェクトA", "total_tasks": 2, "completed_tasks": 0, "completion_rate": 0.0, "overdue_tasks": 0 }, { "project_id": 2, "project_name": "プロジェクトB", "total_tasks": 1, "completed_tasks": 1, "completion_rate": 100.0, "overdue_tasks": 0 } ] }, "id": 1 }
Claude Desktopとの接続
サーバーをClaude Desktopに接続します：
-
設定ファイルの編集：
Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：{ "mcp_servers": [ { "name": "AnalysisServer", "url": "http://localhost:8096", "auth": "none" } ] } -
Claudeでテスト：
Claude Desktopを起動し、プロンプトを入力：プロジェクトの進捗状況を教えてください。レスポンス例：
プロジェクトの進捗状況： - プロジェクトA： - 総タスク数：2 - 完了タスク：0 - 完了率：0% - 期限切れタスク：0 - プロジェクトB： - 総タスク数：1 - 完了タスク：1 - 完了率：100% - 期限切れタスク：0別のプロンプト：
優先すべきプロジェクトは？レスポンス例：
プロジェクトAの完了率が0%で、2つのタスクが未完了です。まずこちらを優先することをおすすめします。
実装のコツと注意点
- パフォーマンス：大量のデータの場合、pandasの処理を最適化（例：インデックス使用）。
- エラーハンドリング：データが空の場合や不正なデータに対応。
- セキュリティ：本番環境では、APIキーを保護し、HTTPSを有効化。
- テスト：テスト用テーブルを作成し、本番データに影響を与えない。
- 拡張性：複雑な分析（例：時系列分析）の場合、SQLクエリをSupabase側で実行。
試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- プロジェクトごとのタスクを期限順にソートする機能を追加。
- 特定のプロジェクトIDを指定して統計を取得するパラメータをサポート。
- 分析結果をCSVファイルにエクスポートするツールを登録。
まとめと次のステップ
この第3章では、Supabaseのデータを活用してデータ分析エージェントを構築しました。pandasを使った統計計算により、AIがプロジェクトの進捗や優先度を提案できるようになりました。
次の第4章では、Supabaseをコミュニティ管理に活用し、リアルタイムチャットボットを構築します。たとえば、Supabaseに保存されたチャットログを基に、AIがユーザーからの質問に即座に回答します。コミュニティAIに興味がある方は、ぜひお楽しみに！
役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！