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 | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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を受信。

インストールコマンド：

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を選択し、以下のスクリプトを追加： 
      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または公式サイト）。
    • ローカルサーバーを公開： 
      ngrok http 8102
    • 生成されたURL（例：https://abc123.ngrok.io）をWebhookスクリプトに設定。

環境変数

第3章の.envファイルに以下を追加：

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がリマインダーを生成する機能を提供します。

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起動：

    ngrok http 8103

    ngrok URL（例：https://abc123.ngrok.io）をAirtableのWebhookスクリプトに設定。

  2. サーバー起動：

    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でリクエストを送信：

    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))

    期待されるレスポンス：

    {
  "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）に以下を追加：

    {
  "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の未来に興味がある方は、ぜひお楽しみに！

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

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?