はじめに

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

Zapier APIトークンの確認：
- 第2章のトークンを使用。
- 権限を確認：Zapsと実行履歴への読み取りアクセス。
Zapierワークフロー準備：
- 第2章のZaps（例：Gmail→Google Sheets、Slack→Trello）を使用。
- 複数のZapsを実行し、成功・失敗の履歴を生成（例：無効なメールアドレスで失敗）。
- 接続アプリ（例：Gmail、Trello、Slack）を確認。
環境変数：
第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が設定済み。

サーバーのテスト

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

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

ワークフロー分析のテスト：
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に接続します：

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

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

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

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