はじめに

第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のセットアップ

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

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

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

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が設定済み。

サーバーのテスト

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

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

Zaps作成のテスト：
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))

期待されるレスポンス：

{
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "zap_id": "12347"
  },
  "id": 1
}

Zapsトグルのテスト：

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

期待されるレスポンス：

{
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "zap_id": "12347",
    "new_status": "off"
  },
  "id": 2
}

Claude Desktopとの接続

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

設定ファイルの編集：
Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：

{
  "mcp_servers": [
    {
      "name": "ZapierTaskServer",
      "url": "http://localhost:8138",
      "auth": "none"
    }
  ]
}

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

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