ZapierでAIを強化する | 第2章：ワークフローを自動化：AIによるZaps操作

## はじめに

第1章では、**Zapier**と**Model Context Protocol（MCP）**の基本を学び、ZapierからZapsや接続アプリのデータを取得するMCPサーバーを構築しました。これにより、AIがワークフローの情報を取得できるようになりました。今回は、この基盤を進化させ、Zapierに新しいZapsを作成したり、既存のZapsを有効化・無効化したりする機能を実装します。これにより、**ワークフローを自動化**するエージェントを構築します。

この第2章では、MCPサーバーを通じてAIがZapsを操作し、特定の条件に基づいて自動化ルールを調整できるようにします。たとえば、AIが「Gmailで特定のメールを受信したらTrelloにタスクを作成する」Zapsを自動で作成できます。コード例とステップごとのガイドで、ワークフロー自動化の可能性を体感しましょう。さあ、始めましょう！

## ワークフロー自動化エージェントとは？

ワークフロー自動化エージェントは、ZapierのZapsを操作し、アプリ間の自動化を効率化するAIです。MCPサーバーを介して、以下のような機能を実現できます：
- **Zaps管理**：新しいZapsの作成、既存Zapsの有効化・無効化。
- **自動化ルール**：特定の条件（例：トリガーやアクションの種類）に基づいてZapsを調整。
- **チーム支援**：ワークフローの最適化やエラー通知を自動化。

### ユースケース
- ワークフロー管理：AIが「エラー通知」メールを受信した際に、自動でTrelloにタスクを作成するZapsを生成。
- エラー対応：AIが頻繁に失敗するZapsを検出し、無効化して代替Zapsを提案。
- コラボレーション強化：AIがZapsの実行結果をSlackに通知。

## 開発環境の準備

第1章の環境を基に、以下の準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**requestsライブラリ**、**Claude Desktop**：第1章と同じ。
- **python-dotenv**：環境変数の管理（既にインストール済み）。
- **Zapierアカウント**：Zaps操作のための設定。

### Zapierのセットアップ
1. **Zapier APIトークンの確認**：
   - 第1章のトークンを使用。
   - 権限を確認：Zapsの作成、編集、有効化・無効化が可能（Developer Platformで設定）。
2. **Zapierワークフロー準備**：
   - テスト用Zapsを追加（例：Gmail→Google Sheets、Slack→Trello）。
   - 接続アプリを準備（例：Gmail、Trello、Slack）。
   - Zapsのトリガーやアクションの設定例を確認（例：Gmailの「New Email」、Trelloの「Create Card」）。
3. **環境変数**：
   第1章の`.env`ファイル（`ZAPIER_API_TOKEN`）を再利用：
   ```bash
   ZAPIER_API_TOKEN=your_zapier_api_token
   ```

## コード例：ワークフロー自動化用MCPサーバー

以下のMCPサーバーは、Zapsデータを取得し、Zapsの作成や有効化・無効化の機能を提供します。

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests

class ZapierTaskServer(MCPServer):
    def __init__(self, host, port, api_token):
        super().__init__(host, port)
        self.api_token = api_token
        self.base_url = "https://api.zapier.com/v1"
        self.headers = {
            "Authorization": f"Bearer {api_token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
        self.register_resource("get_zaps", self.get_zaps)
        self.register_tool("create_zap", self.create_zap)
        self.register_tool("toggle_zap", self.toggle_zap)

    def get_zaps(self, params):
        try:
            url = f"{self.base_url}/zaps"
            query = {"limit": params.get("limit", 10)}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            zaps = response.json()
            zap_list = [
                {
                    "id": zap["id"],
                    "title": zap["title"],
                    "status": zap["status"],
                    "trigger_app": zap["trigger"]["app"]["name"] if zap.get("trigger") else "N/A",
                    "action_app": zap["actions"][0]["app"]["name"] if zap.get("actions") else "N/A",
                    "last_edited": zap["last_edited"]
                }
                for zap in zaps
            ]
            return {"status": "success", "zaps": zap_list}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def create_zap(self, params):
        try:
            title = params.get("title", "")
            trigger_app = params.get("trigger_app", "")
            trigger_event = params.get("trigger_event", "")
            action_app = params.get("action_app", "")
            action_event = params.get("action_event", "")
            if not all([title, trigger_app, trigger_event, action_app, action_event]):
                return {"status": "error", "message": "タイトル、トリガー、アクションが必要です"}
            
            url = f"{self.base_url}/zaps"
            payload = {
                "title": title,
                "status": "on",
                "trigger": {
                    "app": {"name": trigger_app},
                    "event": trigger_event
                },
                "actions": [
                    {
                        "app": {"name": action_app},
                        "event": action_event
                    }
                ]
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            zap = response.json()
            return {"status": "success", "zap_id": zap["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def toggle_zap(self, params):
        try:
            zap_id = params.get("zap_id", "")
            status = params.get("status", "").lower()
            if not zap_id or status not in ["on", "off"]:
                return {"status": "error", "message": "Zaps IDと有効/無効ステータスが必要です"}
            
            url = f"{self.base_url}/zaps/{zap_id}"
            payload = {"status": status}
            response = requests.patch(url, headers=self.headers, json=payload)
            response.raise_for_status()
            zap = response.json()
            return {"status": "success", "zap_id": zap["id"], "new_status": zap["status"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = ZapierTaskServer(
        host="localhost",
        port=8138,
        api_token=os.getenv("ZAPIER_API_TOKEN")
    )
    print("ZapierタスクMCPサーバーを起動中: http://localhost:8138")
    server.start()
```

### コードの説明
- **get_zaps**：第1章から再利用。ユーザーのZapsを取得。
- **create_zap**：新しいZapsを作成。タイトル、トリガーアプリ、トリガーイベント、アクションアプリ、アクションイベントを指定。
- **toggle_zap**：指定したZapsを有効化（on）または無効化（off）。
- **register_resource/tool**：Zaps取得をリソース、Zaps作成・トグルをツールとして登録。

### 前提条件
- Zapier APIトークンにZapsの作成・編集権限がある。
- 接続アプリ（例：Gmail、Trello）がZapierアカウントに設定済み。
- `.env`ファイルに正しい`ZAPIER_API_TOKEN`が設定済み。

## サーバーのテスト

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

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

2. **Zaps作成のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8138"
   payload = {
       "jsonrpc": "2.0",
       "method": "create_zap",
       "params": {
           "title": "Gmail to Trello",
           "trigger_app": "Gmail",
           "trigger_event": "New Email",
           "action_app": "Trello",
           "action_event": "Create Card"
       },
       "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",
       "zap_id": "12347"
     },
     "id": 1
   }
   ```

3. **Zapsトグルのテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "toggle_zap",
       "params": {
           "zap_id": "12347",
           "status": "off"
       },
       "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",
       "zap_id": "12347",
       "new_status": "off"
     },
     "id": 2
   }
   ```

## Claude Desktopとの接続

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

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

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   Gmailで新しいメールを受信したらTrelloにカードを作成するZapsを作成してください。
   ```
   レスポンス例：
   ```
   Zaps「Gmail to Trello」を作成しました（ID: 12347）。
   ```

   別のプロンプト：
   ```
   Zaps ID 12347を無効化してください。
   ```
   レスポンス例：
   ```
   Zaps ID 12347を無効化しました（ステータス：off）。
   ```

## 実装のコツと注意点

- **エラーハンドリング**：無効なトリガーやアクションが指定された場合のエラーを適切に処理。
- **レートリミティング**：Zapier APIの制限（例：100リクエスト/分、プランによる）に注意。
- **セキュリティ**：本番環境では、`auth: none`を避け、トークン認証を導入。
- **テスト**：テスト用Zapsを作成し、本番ワークフローに影響を与えない。
- **拡張性**：複雑なZaps（複数アクション）を作成する場合、詳細なパラメータをサポート。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 特定のキーワード（例：「緊急」）を含むメールに反応するZapsを自動作成。
- Zapsの実行履歴を取得してエラーを検出する機能。
- Zapsのトリガーやアクションを更新するツール。

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

この第2章では、ZapierにZapsを作成し、有効化・無効化するMCPサーバーを構築し、ワークフロー自動化エージェントを実現しました。AIがZapsを操作できるようになり、アプリ間の自動化が効率化されました。

次の第3章では、ZapierのZapsデータを活用して**ワークフロー分析エージェント**を構築します。たとえば、AIがZapsの実行頻度やエラーレートを分析し、ワークフローのインサイトを提供します。ワークフロー分析AIに興味がある方は、ぜひお楽しみに！

---

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