AirtableでAIを強化する | 第3章：データ分析エージェントの構築：インサイトの生成

## はじめに

第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`テーブルを再利用。

インストールコマンド：
```bash
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`テーブルからデータを取得し、プロジェクトごとの統計を計算します。たとえば、タスク完了率や期限超過タスク数を生成します。

```python
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`が設定済み。

## サーバーのテスト

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

1. **サーバー起動**：
   ```bash
   python airtable_analysis_server.py
   ```
   コンソールに「Airtable分析MCPサーバーを起動中: http://localhost:8101」と表示。

2. **統計取得のテスト**：
   Pythonでリクエストを送信：
   ```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))
   ```
   期待されるレスポンス：
   ```json
   {
     "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に接続します：

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "AirtableAnalysisServer",
         "url": "http://localhost:8101",
         "auth": "none"
       }
     ]
   }
   ```

2. **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に興味がある方は、ぜひお楽しみに！

---

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