はじめに
第2章では、Airtableにタスクを追加・更新するMCPサーバーを構築し、タスク管理を自動化するエージェントを実現しました。これにより、AIがユーザーの指示に基づいてアクションを実行できるようになりました。今回は、この基盤を活用して、Airtableのデータを分析し、プロジェクトの進捗や傾向に関するインサイトを生成するデータ分析エージェントを構築します。
この第3章では、AirtableのProjectsとTasksテーブルからデータを取得し、プロジェクトごとの統計（例：タスク完了率、期限超過タスク数）を計算するMCPサーバーを実装します。AIはこれらのインサイトを基に、優先度の高いプロジェクトや遅延リスクを提案できます。コード例とステップごとのガイドで、データ分析AIの構築を体験しましょう。さあ、始めましょう！
データ分析エージェントとは？
データ分析エージェントは、データベースから情報を抽出し、統計や傾向を計算してユーザーにインサイトを提供するAIです。AirtableとMCPを組み合わせることで、以下のような機能を実現できます：
- 進捗分析：プロジェクトごとのタスク数や完了率を計算。
- リスク特定：期限超過タスクや遅延リスクのあるプロジェクトを検出。
- 提案生成：データに基づいて次のアクションを提案（例：優先タスクのフォローアップ）。
ユースケース
- プロジェクト管理：チームの生産性を可視化し、ボトルネックを特定。
- イベント企画：参加者データの傾向を分析し、マーケティング戦略を提案。
- CRM：顧客エンゲージメントを分析し、優先度の高いリードを特定。
開発環境の準備
第2章の環境を基に、以下の追加準備を行います：
- Python 3.8以降、mcpライブラリ、pyairtableライブラリ、Claude Desktop：これまでと同じ。
- pandasライブラリ：データ分析用。
-
Airtableテーブル：第2章の
ProjectsとTasksテーブルを再利用。
インストールコマンド：
pip install pandas
Airtableテーブルの確認
以下のテーブルを使用：
- Projects：プロジェクト名、説明、作成日。
-
Tasks：タスク名、ステータス（
Open、In Progress、Completed）、期限、プロジェクトID。
テストデータを追加（必要に応じて）：
| Name | Status | Due Date | Project ID |
|------------------|------------|------------|------------|
| データ分析 | Completed | 2025-04-17 | プロジェクトB |
| ミーティング準備 | Open | 2025-04-19 | プロジェクトA |
環境変数
第2章の.envファイル（AIRTABLE_TOKEN、AIRTABLE_BASE_ID、AIRTABLE_PROJECTS_TABLE、AIRTABLE_TASKS_TABLE）を再利用。
コード例：データ分析用MCPサーバー
以下のMCPサーバーは、AirtableのProjectsとTasksテーブルからデータを取得し、プロジェクトごとの統計を計算します。たとえば、タスク完了率や期限超過タスク数を生成します。
from mcp import MCPServer
from pyairtable import Table
import os
from dotenv import load_dotenv
import pandas as pd
from datetime import datetime
class AirtableAnalysisServer(MCPServer):
def __init__(self, host, port, airtable_token, base_id, projects_table, tasks_table):
super().__init__(host, port)
self.projects_table = Table(airtable_token, base_id, projects_table)
self.tasks_table = Table(airtable_token, base_id, tasks_table)
self.register_resource("get_project_stats", self.get_project_stats)
def get_project_stats(self, params):
try:
# プロジェクトとタスクを取得
projects = self.projects_table.all()
tasks = self.tasks_table.all()
# pandasでデータ分析
df_projects = pd.DataFrame([
{"id": p["id"], "name": p["fields"].get("Name", "")}
for p in projects
])
df_tasks = pd.DataFrame([
{
"id": t["id"],
"name": t["fields"].get("Name", ""),
"status": t["fields"].get("Status", ""),
"due_date": t["fields"].get("Due Date", ""),
"project_id": t["fields"].get("Project ID", [None])[0]
}
for t in tasks
])
# プロジェクトごとの統計
stats = []
for _, project in df_projects.iterrows():
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") &
(pd.to_datetime(project_tasks["due_date"]) < datetime.utcnow())
])
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 = AirtableAnalysisServer(
host="localhost",
port=8101,
airtable_token=os.getenv("AIRTABLE_TOKEN"),
base_id=os.getenv("AIRTABLE_BASE_ID"),
projects_table=os.getenv("AIRTABLE_PROJECTS_TABLE"),
tasks_table=os.getenv("AIRTABLE_TASKS_TABLE")
)
print("Airtable分析MCPサーバーを起動中: http://localhost:8101")
server.start()
コードの説明
-
get_project_stats：
ProjectsとTasksテーブルからデータを取得し、pandasで分析。 -
統計項目：
- 総タスク数（
total_tasks） - 完了タスク数（
completed_tasks） - 完了率（
completion_rate） - 期限超過タスク数（
overdue_tasks）
- 総タスク数（
- pandas：データフレームを使用して、プロジェクトごとのフィルタリングと計算を効率化。
- register_resource：統計リソースを登録し、AIがアクセス可能に。
前提条件
-
ProjectsとTasksテーブルがAirtableに存在。 -
.envファイルに正しいAIRTABLE_TOKEN、AIRTABLE_BASE_ID、AIRTABLE_PROJECTS_TABLE、AIRTABLE_TASKS_TABLEが設定済み。
サーバーのテスト
サーバーが正しく動作するか確認します：
-
サーバー起動：
python airtable_analysis_server.pyコンソールに「Airtable分析MCPサーバーを起動中: http://localhost:8101」と表示。
-
統計取得のテスト：
Pythonでリクエストを送信：import requests import json url = "http://localhost:8101" 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": "rec123", "project_name": "プロジェクトA", "total_tasks": 3, "completed_tasks": 0, "completion_rate": 0.0, "overdue_tasks": 1 }, { "project_id": "rec456", "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": "AirtableAnalysisServer", "url": "http://localhost:8101", "auth": "none" } ] } -
Claudeでテスト：
Claude Desktopを起動し、プロンプトを入力：プロジェクトの進捗状況を教えてください。レスポンス例：
プロジェクトの進捗状況： - プロジェクトA： - 総タスク数：3 - 完了タスク：0 - 完了率：0% - 期限超過タスク：1 - プロジェクトB： - 総タスク数：1 - 完了タスク：1 - 完了率：100% - 期限超過タスク：0別のプロンプト：
優先すべきプロジェクトは？レスポンス例：
プロジェクトAの完了率が0%で、期限超過タスクが1件あります。まずこちらを優先することをおすすめします。
実装のコツと注意点
- パフォーマンス：大量のデータの場合、Airtable APIのリクエストを最適化（例：フィールド指定）。
- エラーハンドリング：データが空の場合や不正なデータに対応。
- レートリミティング：Airtableの無料枠（5リクエスト/秒）に注意。
- セキュリティ：本番環境では、APIトークンを保護し、HTTPSを有効化。
- 拡張性：複雑な分析の場合、Airtableのビューを活用してデータを事前フィルタリング。
試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 特定のプロジェクトIDを指定して統計を取得するパラメータをサポート。
- タスクを期限順にソートして優先度を提案する機能。
- 分析結果をAirtableの新しいテーブルに保存するツールを登録。
まとめと次のステップ
この第3章では、Airtableのデータを活用してデータ分析エージェントを構築しました。pandasを使った統計計算により、AIがプロジェクトの進捗や優先度を提案できるようになりました。
次の第4章では、Airtableをイベント管理に活用し、リアルタイムイベント管理AIを構築します。たとえば、Airtableに保存されたイベントデータを基に、AIがスケジュール調整やリマインダーを送信します。イベント管理AIに興味がある方は、ぜひお楽しみに！
役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！