LoginSignup
1
1

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

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

Posted at

はじめに

第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）を再利用： 
    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が設定済み。

サーバーのテスト

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

  1. サーバー起動：

    python zapier_task_server.py

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

  2. 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
}

  3. 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に接続します：

  1. 設定ファイルの編集：
    Claude Desktopの設定ファイル（例：claude_desktop_config.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に興味がある方は、ぜひお楽しみに！

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

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

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?