ZapierでAIを強化する | 第3章：ワークフロー分析：AIによるインサイト生成

## はじめに

第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`ファイルに以下を確認：
   ```bash
   ZAPIER_API_TOKEN=your_zapier_api_token
   ```

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

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

```python
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. **サーバー起動**：
   ```bash
   python zapier_analysis_server.py
   ```
   コンソールに「Zapier分析MCPサーバーを起動中: http://localhost:8139」と表示。

2. **ワークフロー分析のテスト**：
   Pythonでリクエストを送信：
   ```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))
   ```
   期待されるレスポンス：
   ```json
   {
     "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`）に以下を追加：
   ```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に興味がある方は、ぜひお楽しみに！

---

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