AirtableでAIを強化する | 第4章：イベント管理AIの構築：リアルタイム連携

## はじめに

第3章では、**Airtable**のデータを活用してデータ分析エージェントを構築し、プロジェクトの進捗や傾向に関するインサイトを生成しました。これにより、AIがタスク完了率や期限超過リスクを提案できるようになりました。今回は、Airtableをイベント管理に活用し、**リアルタイムイベント管理AI**を構築します。このAIは、Airtableに保存されたイベントデータを基に、スケジュール調整やリマインダーを自動化します。

この第4章では、AirtableのWebhook機能とMCPサーバーを統合し、イベントデータの変更（例：新しいイベントの追加）をリアルタイムで捕捉します。AIはこれを利用して、参加者にリマインダーを送信したり、スケジュール衝突を警告したりできます。コード例とステップごとのガイドで、イベント管理AIの構築を体験しましょう。さあ、始めましょう！

## イベント管理AIとは？

イベント管理AIは、イベントのスケジュール、参加者、会場などのデータを管理し、ユーザーやチームの効率を向上させるエージェントです。AirtableのデータベースとMCPサーバーを組み合わせることで、以下のような機能を実現できます：
- **スケジュール管理**：イベントの追加や更新をリアルタイムで処理。
- **リマインダー**：イベント開始前に参加者に通知。
- **衝突検出**：スケジュールの重複を特定し、警告を発信。

### ユースケース
- **イベント企画**：セミナーやワークショップのスケジュールを管理し、参加者にリマインダーを送信。
- **チームミーティング**：会議のスケジュール衝突を防ぎ、議題を整理。
- **コミュニティ管理**：オンラインイベントの参加者データを追跡し、フォローアップを自動化。

## 開発環境の準備

第3章の環境を基に、以下の追加準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**pyairtableライブラリ**、**Claude Desktop**：これまでと同じ。
- **requestsライブラリ**：Webhookのテスト用。
- **Airtableテーブル**：イベントデータを保存する新しいテーブル。
- **ngrok**：ローカルサーバーを公開し、Webhookを受信。

インストールコマンド：
```bash
pip install requests
```

### Airtableテーブルの作成
イベントを管理する`Events`テーブルを作成します。AirtableのUIで新しいテーブルを追加し、以下のフィールドを設定：
- `Title`（テキスト）：イベント名
- `Date`（日付）：開催日時
- `Location`（テキスト）：場所
- `Status`（単一選択）：`Scheduled`、`In Progress`、`Completed`
- `Project ID`（リンク）：`Projects`テーブルへのリンク

テストデータを入力：
| Title             | Date                | Location       | Status     | Project ID |
|-------------------|---------------------|----------------|------------|------------|
| プロジェクトAキックオフ | 2025-04-20 10:00 | オンライン      | Scheduled   | プロジェクトA |
| データ分析ワークショップ | 2025-04-22 14:00 | 会議室B        | Scheduled   | プロジェクトB |

### Airtable Webhookの設定
1. **Webhook有効化**：
   - Airtableの「Automations」タブで新しい自動化を作成。
   - トリガー：`When a record is updated`（`Events`テーブル）。
   - アクション：`Run a script`を選択し、以下のスクリプトを追加：
     ```javascript
     let table = base.getTable("Events");
     let record = input.config().recordId;
     await fetch("YOUR_WEBHOOK_URL", {
         method: "POST",
         headers: {"Content-Type": "application/json"},
         body: JSON.stringify({recordId: record, table: "Events"})
     });
     ```
   - `YOUR_WEBHOOK_URL`は後でngrokから取得。
2. **ngrok設定**：
   - ngrokをインストール（`brew install ngrok`または公式サイト）。
   - ローカルサーバーを公開：
     ```bash
     ngrok http 8102
     ```
   - 生成されたURL（例：`https://abc123.ngrok.io`）をWebhookスクリプトに設定。

### 環境変数
第3章の`.env`ファイルに以下を追加：
```bash
AIRTABLE_TOKEN=your_airtable_token
AIRTABLE_BASE_ID=your_base_id
AIRTABLE_PROJECTS_TABLE=Projects
AIRTABLE_TASKS_TABLE=Tasks
AIRTABLE_EVENTS_TABLE=Events
```

## コード例：イベント管理用MCPサーバー

以下のMCPサーバーは、Airtableの`Events`テーブルからイベントデータを取得し、Webhook経由で変更を捕捉します。さらに、AIがリマインダーを生成する機能を提供します。

```python
from mcp import MCPServer
from pyairtable import Table
import os
from dotenv import load_dotenv
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading

class AirtableEventServer(MCPServer):
    def __init__(self, host, port, airtable_token, base_id, projects_table, events_table):
        super().__init__(host, port)
        self.projects_table = Table(airtable_token, base_id, projects_table)
        self.events_table = Table(airtable_token, base_id, events_table)
        self.latest_event = None
        self.register_resource("get_events", self.get_events)
        self.register_resource("get_latest_event", self.get_latest_event)
        self.start_webhook_server()

    def get_events(self, params):
        try:
            project_name = params.get("project_name", "")
            if project_name:
                projects = self.projects_table.all(formula=f"{{Name}}='{project_name}'")
                if not projects:
                    return {"status": "error", "message": f"プロジェクト '{project_name}' が見つかりません"}
                project_id = projects[0]["id"]
                records = self.events_table.all(formula=f"FIND('{project_id}', ARRAYJOIN({{Project ID}}))")
            else:
                records = self.events_table.all()

            events = [
                {
                    "id": record["id"],
                    "title": record["fields"].get("Title", ""),
                    "date": record["fields"].get("Date", ""),
                    "location": record["fields"].get("Location", ""),
                    "status": record["fields"].get("Status", ""),
                    "project_id": record["fields"].get("Project ID", [None])[0]
                }
                for record in records
            ]
            return {"status": "success", "events": events}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_latest_event(self, params):
        try:
            if self.latest_event:
                record = self.events_table.get(self.latest_event["recordId"])
                event = {
                    "id": record["id"],
                    "title": record["fields"].get("Title", ""),
                    "date": record["fields"].get("Date", ""),
                    "location": record["fields"].get("Location", ""),
                    "status": record["fields"].get("Status", ""),
                    "project_id": record["fields"].get("Project ID", [None])[0]
                }
                return {"status": "success", "event": event}
            return {"status": "success", "event": None, "message": "イベントなし"}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def start_webhook_server(self):
        class WebhookHandler(BaseHTTPRequestHandler):
            def do_POST(self):
                content_length = int(self.headers["Content-Length"])
                post_data = self.rfile.read(content_length)
                self.server.parent.latest_event = json.loads(post_data.decode("utf-8"))
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"Webhook received")

        server = HTTPServer(("localhost", 8103), WebhookHandler)
        server.parent = self
        threading.Thread(target=server.serve_forever, daemon=True).start()
        print("Webhookサーバーを起動中: http://localhost:8103")

if __name__ == "__main__":
    load_dotenv()
    server = AirtableEventServer(
        host="localhost",
        port=8102,
        airtable_token=os.getenv("AIRTABLE_TOKEN"),
        base_id=os.getenv("AIRTABLE_BASE_ID"),
        projects_table=os.getenv("AIRTABLE_PROJECTS_TABLE"),
        events_table=os.getenv("AIRTABLE_EVENTS_TABLE")
    )
    print("AirtableイベントMCPサーバーを起動中: http://localhost:8102")
    server.start()
```

### コードの説明
- **get_events**：`Events`テーブルからイベントを取得。プロジェクト名でフィルタリング可能。
- **get_latest_event**：Webhookで捕捉した最新のイベント変更を返却。
- **start_webhook_server**：ローカルWebhookサーバーを起動し、Airtableからの更新を受信。
- **WebhookHandler**：AirtableのWebhookリクエストを処理し、イベントデータを保存。

### 前提条件
- `Projects`と`Events`テーブルがAirtableに存在。
- AirtableのWebhookが設定済みで、ngrok URLが正しく登録。
- `.env`ファイルに正しい`AIRTABLE_TOKEN`、`AIRTABLE_BASE_ID`、`AIRTABLE_PROJECTS_TABLE`、`AIRTABLE_EVENTS_TABLE`が設定済み。

## サーバーのテスト

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

1. **ngrok起動**：
   ```bash
   ngrok http 8103
   ```
   ngrok URL（例：`https://abc123.ngrok.io`）をAirtableのWebhookスクリプトに設定。

2. **サーバー起動**：
   ```bash
   python airtable_event_server.py
   ```
   コンソールに「AirtableイベントMCPサーバーを起動中: http://localhost:8102」と「Webhookサーバーを起動中: http://localhost:8103」が表示。

3. **Webhookのテスト**：
   - AirtableのUIで`Events`テーブルに新しいイベントを追加：
     | Title             | Date                | Location       | Status     | Project ID |
     |-------------------|---------------------|----------------|------------|------------|
     | チームミーティング | 2025-04-23 09:00 | オンライン      | Scheduled   | プロジェクトA |
   - サーバーのコンソールにWebhook受信ログが表示。

4. **最新イベント取得のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8102"
   payload = {
       "jsonrpc": "2.0",
       "method": "get_latest_event",
       "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",
       "event": {
         "id": "rec789",
         "title": "チームミーティング",
         "date": "2025-04-23T09:00:00",
         "location": "オンライン",
         "status": "Scheduled",
         "project_id": "rec123"
       }
     },
     "id": 1
   }
   ```

## Claude Desktopとの接続

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

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

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   最新のイベントを教えてください。
   ```
   レスポンス例：
   ```
   最新のイベント：
   - タイトル：チームミーティング
   - 日時：2025-04-23 09:00
   - 場所：オンライン
   - ステータス：予定済み
   - プロジェクト：プロジェクトA
   ```

## 実装のコツと注意点

- **Webhookの安定性**：ngrokの無料枠ではセッションが切れる場合がある。本番環境では固定URL（例：AWS Lambda）を使用。
- **レートリミティング**：AirtableのAPI制限（無料枠：5リクエスト/秒）に注意。
- **セキュリティ**：Webhookエンドポイントに認証トークンを追加し、不正アクセスを防止。
- **テスト**：テスト用ベースを作成し、本番データに影響を与えない。
- **拡張性**：大量のイベントを処理する場合、キャッシュやキュー（例：Redis）を検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- イベントの開始24時間前にリマインダーメッセージを生成するツール。
- スケジュール衝突を検出する機能（例：同じ時間帯のイベントを警告）。
- Webhookでキャプチャしたイベントを外部ツール（例：Slack）に送信。

## まとめと次のステップ

この第4章では、AirtableのWebhookを活用してリアルタイムイベント管理AIを構築しました。イベントデータの変更を即座に捕捉し、AIがスケジュール管理やリマインダー生成を支援できるようになりました。

次の第5章では、MCPサーバーの**最適化とコミュニティへの貢献**に焦点を当てます。サーバーのパフォーマンス向上、セキュリティ強化、そしてAirtable用MCPサーバーをオープンソースとして共有する方法を学びます。コミュニティAIの未来に興味がある方は、ぜひお楽しみに！

---

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