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を強化する | 第2章：タスクを自動化：AIによるアクション

Posted at

はじめに

第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テーブル名を指定）：

AIRTABLE_TOKEN=your_airtable_token
AIRTABLE_BASE_ID=your_base_id
AIRTABLE_PROJECTS_TABLE=Projects
AIRTABLE_TASKS_TABLE=Tasks

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

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

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. サーバー起動：

    python airtable_task_server.py

    コンソールに「AirtableタスクMCPサーバーを起動中: http://localhost:8100」と表示。

  2. タスク追加のテスト：
    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))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "task_id": "rec789"
  },
  "id": 1
}

  3. タスク取得のテスト：

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

    期待されるレスポンス：

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

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

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

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?