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を強化する | 第3章：ワークフロー分析：AIによるインサイト生成

Posted at

はじめに

第2章では、Zapierに新しいZapsを作成し、有効化・無効化するMCPサーバーを構築し、ワークフロー自動化エージェントを実現しました。これにより、AIがアプリ間の自動化を操作できるようになり、効率が向上しました。今回は、この基盤を活用して、ZapierのZapsデータを解析するワークフロー分析エージェントを構築します。

この第3章では、Zapsの実行履歴や接続アプリのデータを分析し、ワークフローの効率、エラーレート、ボトルネックに関するインサイトを生成します。たとえば、AIが「特定のZapsが頻繁に失敗している」ことを検出したり、「過負荷のアプリ接続」を指摘したりできます。コード例とステップごとのガイドで、ワークフロー分析AIの構築を体験しましょう。さあ、始めましょう！

ワークフロー分析エージェントとは？

ワークフロー分析エージェントは、ZapierのZapsや接続アプリのデータを解析し、自動化ワークフローのパフォーマンスを評価するAIです。MCPサーバーを介して、以下のような機能を実現できます：

  • 実行分析：Zapsの実行頻度や成功率を評価。
  • エラー検出：失敗したZapsや問題のあるアプリ接続を特定。
  • 最適化提案：非効率なワークフローやボトルネックを改善する提案。

ユースケース

  • ワークフロー管理：頻繁に失敗するZapsを特定し、代替アクションを提案。
  • パフォーマンス監視：アプリ接続の使用率やエラーレートを分析。
  • プロセス改善：過負荷のトリガーやアクションを検出し、ワークフローを最適化。

開発環境の準備

第2章の環境を基に、以下の準備を行います：

  • Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：これまでと同じ。
  • python-dotenv：環境変数の管理（既にインストール済み）。
  • Zapierアカウント：分析用のZapsと実行履歴。

Zapierのセットアップ

  1. Zapier APIトークンの確認：
    • 第2章のトークンを使用。
    • 権限を確認：Zapsと実行履歴への読み取りアクセス。
  2. Zapierワークフロー準備：
    • 第2章のZaps（例：Gmail→Google Sheets、Slack→Trello）を使用。
    • 複数のZapsを実行し、成功・失敗の履歴を生成（例：無効なメールアドレスで失敗）。
    • 接続アプリ（例：Gmail、Trello、Slack）を確認。
  3. 環境変数：
    第2章の.envファイルに以下を確認： 
    ZAPIER_API_TOKEN=your_zapier_api_token

コード例：ワークフロー分析用MCPサーバー

以下のMCPサーバーは、ZapierのZapsと実行履歴を取得し、ワークフローのパフォーマンスを分析します。

from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from collections import Counter
from datetime import datetime, timedelta

class ZapierAnalysisServer(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"
        }
        self.register_resource("analyze_workflow", self.analyze_workflow)

    def get_zaps(self):
        try:
            url = f"{self.base_url}/zaps"
            query = {"limit": 100}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_execution_history(self, zap_id):
        try:
            url = f"{self.base_url}/zaps/{zap_id}/history"
            query = {"limit": 50}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def analyze_workflow(self, params):
        try:
            zaps = self.get_zaps()
            if isinstance(zaps, dict) and "status" in zaps:
                return zaps

            # Zaps分析
            zap_status_counts = Counter(zap["status"] for zap in zaps)
            app_usage = Counter()
            for zap in zaps:
                if zap.get("trigger"):
                    app_usage[zap["trigger"]["app"]["name"]] += 1
                if zap.get("actions"):
                    for action in zap["actions"]:
                        app_usage[action["app"]["name"]] += 1

            # エラー分析
            failed_zaps = []
            for zap in zaps:
                history = self.get_execution_history(zap["id"])
                if isinstance(history, dict) and "status" in history:
                    continue
                failed_executions = [h for h in history if h["status"] == "failed"]
                if failed_executions:
                    failure_rate = len(failed_executions) / len(history) * 100 if history else 0
                    failed_zaps.append({
                        "id": zap["id"],
                        "title": zap["title"],
                        "failure_rate": f"{failure_rate:.2f}%",
                        "last_failed": max((h["timestamp"] for h in failed_executions), default="N/A")
                    })

            # 最近の実行頻度
            recent_executions = 0
            one_day_ago = datetime.now() - timedelta(days=1)
            for zap in zaps:
                history = self.get_execution_history(zap["id"])
                if isinstance(history, dict) and "status" in history:
                    continue
                recent_executions += len([h for h in history 
                                        if datetime.fromisoformat(h["timestamp"].replace("Z", "+00:00")) > one_day_ago])

            return {
                "status": "success",
                "analysis": {
                    "zap_count": len(zaps),
                    "status_distribution": [{"status": s, "count": c} for s, c in zap_status_counts.items()],
                    "app_usage": [{"app": a, "count": c} for a, c in app_usage.items()],
                    "failed_zaps": failed_zaps,
                    "recent_execution_count": recent_executions
                }
            }
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = ZapierAnalysisServer(
        host="localhost",
        port=8139,
        api_token=os.getenv("ZAPIER_API_TOKEN")
    )
    print("Zapier分析MCPサーバーを起動中: http://localhost:8139")
    server.start()

コードの説明

  • get_zaps：ユーザーのZapsを取得（最大100件）。
  • get_execution_history：指定したZapsの実行履歴を取得（最大50件）。
  • analyze_workflow：Zapsと実行履歴を解析し、以下のインサイトを生成：
    • Zaps総数：取得したZapsの数。
    • ステータス分布：各ステータス（on、off）のZaps数。
    • アプリ使用状況：トリガーおよびアクションで使用されているアプリの頻度。
    • 失敗Zaps：失敗率と最後の失敗日時を含むZapsリスト。
    • 最近の実行頻度：過去24時間の実行回数。
  • register_resource：ワークフロー分析をリソースとして登録。

前提条件

  • 複数のZapsが稼働し、成功・失敗の実行履歴が存在。
  • Zapier APIトークンにZapsと実行履歴への読み取り権限がある。
  • .envファイルに正しいZAPIER_API_TOKENが設定済み。

サーバーのテスト

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

  1. サーバー起動：

    python zapier_analysis_server.py

    コンソールに「Zapier分析MCPサーバーを起動中: http://localhost:8139」と表示。

  2. ワークフロー分析のテスト：
    Pythonでリクエストを送信：

    import requests
import json

url = "http://localhost:8139"
payload = {
    "jsonrpc": "2.0",
    "method": "analyze_workflow",
    "params": {},
    "id": 1
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "analysis": {
      "zap_count": 20,
      "status_distribution": [
        {"status": "on", "count": 15},
        {"status": "off", "count": 5}
      ],
      "app_usage": [
        {"app": "Gmail", "count": 10},
        {"app": "Trello", "count": 8},
        {"app": "Slack", "count": 5}
      ],
      "failed_zaps": [
        {
          "id": "12345",
          "title": "Gmail to Sheets",
          "failure_rate": "20.00%",
          "last_failed": "2025-04-22T12:00:00Z"
        }
      ],
      "recent_execution_count": 50
    }
  },
  "id": 1
}

Claude Desktopとの接続

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

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

    {
  "mcp_servers": [
    {
      "name": "ZapierAnalysisServer",
      "url": "http://localhost:8139",
      "auth": "none"
    }
  ]
}

  2. Claudeでテスト：
    Claude Desktopを起動し、プロンプトを入力：

    ワークフローの分析結果を教えてください。

    レスポンス例：

    ワークフロー分析結果：
- Zaps総数：20
- ステータス分布：on（15）、off（5）
- アプリ使用状況：Gmail（10）、Trello（8）、Slack（5）
- 失敗Zaps：Gmail to Sheets（ID: 12345、失敗率：20.00%、最終失敗：2025-04-22）
- 最近の実行回数（過去24時間）：50

実装のコツと注意点

  • データ品質：Zapsや実行履歴が少ない場合、分析結果が制限される。十分なデータ（例：20件以上のZaps）を用意。
  • レートリミティング：Zapier APIの制限（例：100リクエスト/分、プランによる）に注意。
  • セキュリティ：本番環境では、auth: noneを避け、トークン認証を導入。
  • テスト：テスト用Zapsを作成し、本番ワークフローに影響を与えない。
  • 拡張性：大量のデータを処理する場合、キャッシュ（例：Redis）やNLPライブラリで履歴分析を強化。

試してみよう：挑戦課題

以下の機能を追加して、エージェントを強化してみてください：

  • 特定のアプリ（例：Gmail）に関連するZapsだけを分析するフィルター。
  • 失敗したZapsのエラーメッセージを分析し、原因を特定。
  • 分析結果を新しいZapsとしてZapierに投稿するツール。

まとめと次のステップ

この第3章では、ZapierのZapsと実行履歴を活用してワークフロー分析エージェントを構築しました。実行頻度やエラーレートを分析することで、AIがワークフローのインサイトを提供し、プロセス改善を支援できるようになりました。

次の第4章では、ZapierのWebhookを活用してリアルタイム管理AIを構築します。たとえば、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?