AirtableでAIを強化する | 第2章：タスクを自動化：AIによるアクション

## はじめに

第1章では、**Airtable**と**Model Context Protocol（MCP）**の基本を学び、Airtableの`Projects`テーブルからデータを取得するMCPサーバーを構築しました。これにより、AIがプロジェクト情報を取得し、データ管理の基礎を体験できました。今回は、この基盤を進化させ、Airtableにレコードを追加・更新する機能を実装することで、**タスク管理を自動化**するエージェントを構築します。

この第2章では、MCPサーバーを通じてAIが新しいタスクをAirtableに追加したり、既存のタスクのステータスを更新したりできるようにします。たとえば、ユーザーが「プロジェクトAにタスクを追加して」と依頼すると、AIがAirtableにレコードを挿入します。コード例とステップごとのガイドで、タスク自動化の可能性を体感しましょう。さあ、始めましょう！

## タスク自動化エージェントとは？

タスク自動化エージェントは、AIがデータベース（Airtable）にアクションを実行し、ユーザーの指示に基づいてタスクを管理するシステムです。MCPサーバーを介して、以下のような機能を実現できます：
- **レコード追加**：新しいタスクやプロジェクトをAirtableに作成。
- **レコード更新**：タスクのステータスや期限を変更。
- **自動化ルール**：特定の条件（例：期限超過）に基づいてアクションをトリガー。

### ユースケース
- プロジェクト管理：AIがタスクを自動追加し、進捗を追跡。
- イベント企画：AIが参加者リストを更新し、リマインダーを設定。
- CRM：AIが顧客フォローアップタスクを自動生成。

## 開発環境の準備

第1章の環境を基に、以下の準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**pyairtableライブラリ**、**Claude Desktop**：第1章と同じ。
- **Airtableテーブル**：タスク管理用の新しいテーブルを作成。
- **python-dotenv**：環境変数の管理（既にインストール済み）。

### Airtableテーブルの拡張
第1章の`Projects`テーブルに加え、タスクを管理する`Tasks`テーブルを作成します。AirtableのUIで新しいテーブルを追加し、以下のフィールドを設定：
- `Name`（テキスト）：タスク名
- `Status`（単一選択）：`Open`、`In Progress`、`Completed`
- `Due Date`（日付）：期限
- `Project ID`（リンク）：`Projects`テーブルへのリンク
テストデータを手動で入力：
| Name             | Status     | Due Date   | Project ID |
|------------------|------------|------------|------------|
| ドキュメント作成 | In Progress | 2025-04-18 | プロジェクトA |
| コードレビュー   | Open       | 2025-04-20 | プロジェクトA |

### 環境変数
第1章の`.env`ファイルに以下を追加（`Tasks`テーブル名を指定）：
```bash
AIRTABLE_TOKEN=your_airtable_token
AIRTABLE_BASE_ID=your_base_id
AIRTABLE_PROJECTS_TABLE=Projects
AIRTABLE_TASKS_TABLE=Tasks
```

## コード例：タスク自動化用MCPサーバー

以下のMCPサーバーは、Airtableの`Tasks`テーブルに新しいタスクを追加し、既存のタスクのステータスを更新する機能を提供します。

```python
from mcp import MCPServer
from pyairtable import Table
import os
from dotenv import load_dotenv

class AirtableTaskServer(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_tool("add_task", self.add_task)
        self.register_tool("update_task_status", self.update_task_status)
        self.register_resource("get_tasks", self.get_tasks)

    def add_task(self, params):
        try:
            name = params.get("name", "")
            status = params.get("status", "Open")
            due_date = params.get("due_date", "")
            project_name = params.get("project_name", "")
            if not name or not project_name:
                return {"status": "error", "message": "タスク名とプロジェクト名が必要です"}

            # プロジェクトIDを検索
            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"]

            # タスクを追加
            task = {
                "Name": name,
                "Status": status,
                "Due Date": due_date,
                "Project ID": [project_id]
            }
            response = self.tasks_table.create(task)
            return {"status": "success", "task_id": response["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def update_task_status(self, params):
        try:
            task_id = params.get("task_id", "")
            status = params.get("status", "")
            if not task_id or not status:
                return {"status": "error", "message": "タスクIDとステータスが必要です"}

            # ステータスを更新
            update = {"Status": status}
            response = self.tasks_table.update(task_id, update)
            return {"status": "success", "task_id": response["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_tasks(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.tasks_table.all(formula=f"FIND('{project_id}', ARRAYJOIN({{Project ID}}))")
            else:
                records = self.tasks_table.all()

            tasks = [
                {
                    "id": record["id"],
                    "name": record["fields"].get("Name", ""),
                    "status": record["fields"].get("Status", ""),
                    "due_date": record["fields"].get("Due Date", ""),
                    "project_id": record["fields"].get("Project ID", [None])[0]
                }
                for record in records
            ]
            return {"status": "success", "tasks": tasks}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = AirtableTaskServer(
        host="localhost",
        port=8100,
        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:8100")
    server.start()
```

### コードの説明
- **add_task**：新しいタスクを`Tasks`テーブルに追加。プロジェクト名を基に`Project ID`をリンク。
- **update_task_status**：指定したタスクのステータスを更新（例：`Open`から`In Progress`）。
- **get_tasks**：プロジェクト名でフィルタリングしてタスクを取得。
- **register_tool/resource**：タスク追加と更新をツール、取得をリソースとして登録。

### 前提条件
- Airtableベースに`Projects`と`Tasks`テーブルが存在。
- `.env`ファイルに正しい`AIRTABLE_TOKEN`、`AIRTABLE_BASE_ID`、`AIRTABLE_PROJECTS_TABLE`、`AIRTABLE_TASKS_TABLE`が設定済み。

## サーバーのテスト

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

1. **サーバー起動**：
   ```bash
   python airtable_task_server.py
   ```
   コンソールに「AirtableタスクMCPサーバーを起動中: http://localhost:8100」と表示。

2. **タスク追加のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8100"
   payload = {
       "jsonrpc": "2.0",
       "method": "add_task",
       "params": {
           "name": "テストタスク",
           "status": "Open",
           "due_date": "2025-04-22",
           "project_name": "プロジェクトA"
       },
       "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",
       "task_id": "rec789"
     },
     "id": 1
   }
   ```

3. **タスク取得のテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "get_tasks",
       "params": {"project_name": "プロジェクトA"},
       "id": 2
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "tasks": [
         {
           "id": "rec123",
           "name": "ドキュメント作成",
           "status": "In Progress",
           "due_date": "2025-04-18",
           "project_id": "rec456"
         },
         {
           "id": "rec789",
           "name": "テストタスク",
           "status": "Open",
           "due_date": "2025-04-22",
           "project_id": "rec456"
         }
       ]
     },
     "id": 2
   }
   ```

## Claude Desktopとの接続

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

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

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   プロジェクトAのタスク一覧を教えてください。
   ```
   レスポンス例：
   ```
   プロジェクトAのタスク：
   - ドキュメント作成（進行中、期限：2025-04-18）
   - テストタスク（未着手、期限：2025-04-22）
   ```

   別のプロンプト：
   ```
   プロジェクトAに「データ分析」というタスクを追加してください。期限は2025-04-25です。
   ```
   レスポンス例：
   ```
   「データ分析」タスクをプロジェクトAに追加しました（期限：2025-04-25）。
   ```

## 実装のコツと注意点

- **エラーハンドリング**：無効なプロジェクト名やタスクIDを適切に処理。
- **レートリミティング**：AirtableのAPI制限（無料枠：5リクエスト/秒）に注意。
- **セキュリティ**：本番環境では、`auth: none`を避け、OAuthやトークン認証を導入。
- **テスト**：テスト用ベースを作成し、本番データに影響を与えない。
- **拡張性**：大量のタスクを扱う場合、キャッシュ（例：Redis）を検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- タスクの削除機能（`delete_task`ツール）を追加。
- 期限が近いタスクを自動で通知する機能を登録。
- タスクの優先度フィールドを追加し、AIが優先順位を提案。

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

この第2章では、Airtableにタスクを追加・更新するMCPサーバーを構築し、タスク管理を自動化するエージェントを実現しました。AIがユーザーの指示に基づいてアクションを実行できるようになり、生産性が向上しました。

次の第3章では、Airtableのデータを活用して**データ分析エージェント**を構築します。たとえば、プロジェクトごとのタスク進捗や傾向を分析し、AIがインサイトを提供します。データ分析AIに興味がある方は、ぜひお楽しみに！

---

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