ZapierでAIを強化する | 第1章：Zapsデータの第一歩

## はじめに

**Zapier**は、Gmail、Slack、Trello、Google Sheetsなど、数千のアプリを接続してワークフローを自動化するプラットフォームです。**Zaps**と呼ばれる自動化ルールを使い、コードなしでタスクを効率化できます。Zapierは強力なREST APIとWebhookを提供し、Zapsのデータ取得や操作が可能です。一方、**Model Context Protocol（MCP）**は、AI（例：Claude）を外部データやツールと標準化された方法で接続するプロトコルです。このシリーズでは、ZapierとMCPを統合し、Zapsデータを活用するエージェントAIを構築する方法を学びます。たとえば、AIがZapsの実行履歴を分析したり、新しいZapsを提案したりできます。

第1章では、ZapierとMCPの基本を理解し、ZapierからZapsや接続アプリのデータを取得するシンプルなMCPサーバーを構築します。Claude Desktopと接続して、ワークフロー自動化の第一歩を踏み出しましょう。

## ZapierとMCPの概要

### Zapierとは？
Zapierは、以下のような特徴を持つ自動化プラットフォームです：
- **Zaps**：トリガー（例：新しいメール受信）とアクション（例：Slackに通知）を組み合わせた自動化ルール。
- **強力なAPI**：REST APIでZaps、トリガー、アクション、アプリ接続のデータにアクセス。
- **Webhookサポート**：リアルタイムでイベント（例：Zaps実行）を検知。
- **多用途**：個人タスクから企業ワークフローまで対応。

### MCPの役割
MCPは、AIにZapierのデータを提供する橋渡し役です。MCPサーバーを通じて、AIは以下を行えます：
- **データ取得**：Zapsの設定、実行履歴、接続アプリ情報を取得。
- **アクション実行**：新しいZapsの作成、トリガーやアクションの更新。
- **分析**：ワークフローの効率やボトルネックを評価。

### なぜZapierとMCPを組み合わせる？
この組み合わせは、以下のようなシナリオで強力です：
- 自動化最適化：AIが頻繁に失敗するZapsを検出し、代替アクションを提案。
- ワークフロー分析：AIがZapsの実行頻度やパフォーマンスを評価。
- 通知強化：AIが重要なイベント（例：Zapsエラー）をSlackやメールで通知。

## 開発環境の準備

MCPサーバーとZapierを統合するには、以下の環境を準備します：
- **Python 3.8以降**：サーバー開発用。
- **mcpライブラリ**：JSON-RPC通信を処理（仮定のライブラリ、公式ドキュメントを参照）。
- **requestsライブラリ**：Zapier APIとの通信用。
- **Claude Desktop**：MCP対応のAIクライアント。
- **テキストエディタ**：VS Code推奨。

インストールコマンド：
```bash
pip install mcp jsonrpcserver requests python-dotenv
```

### Zapierのセットアップ
1. **Zapier APIトークンの作成**：
   - Zapierアカウントにログインし、[Developer Platform](https://developer.zapier.com/)にアクセス。
   - 新しいアプリを作成し、APIトークン（例：`your_zapier_api_token`）を取得。
   - トークンを保存。
2. **Zapierワークフロー準備**：
   - テスト用Zapsを作成（例：Gmailで新しいメール受信→Google Sheetsに記録）。
   - 複数のアプリを接続（例：Slack、Trello、Google Sheets）。
   - Zapsを有効化し、実行履歴を生成。
3. **環境変数の設定**：
   `.env`ファイルに以下を追加：
   ```bash
   ZAPIER_API_TOKEN=your_zapier_api_token
   ```

## コード例：Zapier用MCPサーバー

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests

class ZapierServer(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("get_zaps", self.get_zaps)
        self.register_resource("get_apps", self.get_apps)

    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 get_apps(self, params):
        try:
            url = f"{self.base_url}/apps"
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            apps = response.json()
            app_list = [
                {
                    "id": app["id"],
                    "name": app["name"],
                    "description": app["description"],
                    "connected": app["connected"]
                }
                for app in apps
            ]
            return {"status": "success", "apps": app_list}
        except Exception as e:
            return {"status": "error", "message": str(e)}

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

### コードの説明
- **requests**：Zapier APIにGETリクエストを送信し、Zapsや接続アプリを取得。
- **get_zaps**：ユーザーのZapsを取得（ID、タイトル、ステータス、トリガーアプリ、アクションアプリ、編集日）。
- **get_apps**：接続済みのアプリを取得（ID、名前、説明、接続状態）。
- **register_resource**：Zapsとアプリ取得をリソースとして登録。
- **start()**：サーバーを起動。

### 前提条件
- Zapier APIトークンが生成済み。
- テスト用Zapsが作成済みで、複数のアプリが接続済み。
- `.env`ファイルに正しい`ZAPIER_API_TOKEN`が設定済み。
- ユーザーにZapierアカウントへのアクセス権限がある。

## サーバーのテスト

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

1. **サーバー起動**：
   ```bash
   python zapier_server.py
   ```
   コンソールに「Zapier MCPサーバーを起動中: http://localhost:8137」と表示。

2. **Zaps取得のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8137"
   payload = {
       "jsonrpc": "2.0",
       "method": "get_zaps",
       "params": {"limit": 5},
       "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",
       "zaps": [
         {
           "id": "12345",
           "title": "Gmail to Sheets",
           "status": "on",
           "trigger_app": "Gmail",
           "action_app": "Google Sheets",
           "last_edited": "2025-04-22T10:00:00Z"
         },
         {
           "id": "12346",
           "title": "Slack to Trello",
           "status": "on",
           "trigger_app": "Slack",
           "action_app": "Trello",
           "last_edited": "2025-04-22T11:00:00Z"
         }
       ]
     },
     "id": 1
   }
   ```

3. **アプリ取得のテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "get_apps",
       "params": {},
       "id": 2
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "apps": [
         {
           "id": "1001",
           "name": "Gmail",
           "description": "Email by Google",
           "connected": true
         },
         {
           "id": "1002",
           "name": "Slack",
           "description": "Team communication",
           "connected": true
         }
       ]
     },
     "id": 2
   }
   ```

## Claude Desktopとの接続

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

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "ZapierServer",
         "url": "http://localhost:8137",
         "auth": "none"
       }
     ]
   }
   ```

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   アクティブなZapsを教えてください。
   ```
   レスポンス例：
   ```
   アクティブなZaps：
   - ID: 12345、タイトル：Gmail to Sheets（トリガー：Gmail、アクション：Google Sheets、編集日：2025-04-22）
   - ID: 12346、タイトル：Slack to Trello（トリガー：Slack、アクション：Trello、編集日：2025-04-22）
   ```

   別のプロンプト：
   ```
   接続済みのアプリを教えてください。
   ```
   レスポンス例：
   ```
   接続済みのアプリ：
   - Gmail（ID: 1001、説明：Email by Google）
   - Slack（ID: 1002、説明：Team communication）
   ```

## 次のステップ

このMCPサーバーは、ZapierとAIの連携の基礎です。次の第2章では、Zapierに新しいZapsを作成したり、トリガーやアクションを操作したりする機能を実装し、ワークフローを自動化するエージェントを構築します。たとえば、AIが特定の条件に基づいてZapsを有効化できます。

## まとめ

この第1章では、ZapierとMCPの基本を学び、シンプルなMCPサーバーを構築してZapsと接続アプリのデータを取得しました。Zapierの強力なAPIとMCPの柔軟性を組み合わせることで、AIをワークフロー自動化の強力なアシスタントに変えられます。次章以降では、Zaps管理、ワークフロー分析、リアルタイム管理へと進化させます。

---

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