Dify APIを使ったワークフロー実行するPythonスクリプト

Pythonプログラムから、自分のローカルのDify APIを使ってワークフローを実行するスクリプトが動作してくれたので、メモをしておきます。

```python
import requests
import json
import argparse

API_KEY = "app-XXXXXXXXXXXXXue"  # DifyのAPIキーを入れてください
BASE_URL = "http://localhost/v1"

def run_workflow(inputs, response_mode, user):
    url = f"{BASE_URL}/workflows/run"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "inputs": {"input_text": inputs["input_text"]},
        "response_mode": response_mode,
        "user": user
    }
    print(f"Sending request to {url} with data: {json.dumps(data, ensure_ascii=False)}")  # リクエストデータを出力
    response = requests.post(url, headers=headers, data=json.dumps(data))

    print(f"Received response with status code {response.status_code}")  # ステータスコードを出力

    if response.status_code == 200:
        if response_mode == "blocking":
            result = response.json()
            print(f"Response JSON: {result}")  # レスポンス全体を出力
            if "data" in result and "outputs" in result["data"] and "text" in result["data"]["outputs"]:
                return result["data"]["outputs"]["text"]
            else:
                print("Error: 'text' not found in the API response.")
        elif response_mode == "streaming":
            for chunk in response.iter_content(chunk_size=None):
                if chunk:
                    print(chunk.decode())
    else:
        print(f"Request failed with status code {response.status_code}")
        try:
            print(response.json())
        except json.JSONDecodeError:
            print(response.text)

def stop_workflow(task_id, user):
    url = f"{BASE_URL}/workflows/{task_id}/stop"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    data = {"user": user}
    response = requests.post(url, headers=headers, data=json.dumps(data))

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Request failed with status code {response.status_code}")

def get_parameters(user):
    url = f"{BASE_URL}/parameters?user={user}"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Request failed with status code {response.status_code}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Run workflow")
    parser.add_argument("--input", type=str, required=True, help="Input file path")
    parser.add_argument("--response_mode", type=str, help="Response mode (blocking or streaming)")
    parser.add_argument("--user", type=str, default="example_user", help="User identifier")

    args = parser.parse_args()

    with open(args.input, 'r', encoding='utf-8') as file:
        input_text = file.read().strip()

    inputs = {"input_text": input_text}
    response_mode = args.response_mode if args.response_mode else "blocking"
    user = args.user

    result = run_workflow(inputs, response_mode, user)
    if result:
        print(result)
```


以下、各部分について詳しく説明します。

### インポートと定数の定義

```python
import requests
import json
import argparse
```

- `requests`: HTTPリクエストを送るためのライブラリ。
- `json`: JSONデータを操作するためのライブラリ。
- `argparse`: コマンドライン引数を解析するためのライブラリ。

```python
API_KEY = "app-XXXXXXXXXXXXXue"  # DifyのAPIキーを入れてください
BASE_URL = "http://localhost/v1"
```

- `API_KEY`: Dify APIを使用するための認証キー。
- `BASE_URL`: APIのベースURL。

### ワークフローの実行関数

```python
def run_workflow(inputs, response_mode, user):
    url = f"{BASE_URL}/workflows/run"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "inputs": {"input_text": inputs["input_text"]},
        "response_mode": response_mode,
        "user": user
    }
    print(f"Sending request to {url} with data: {json.dumps(data, ensure_ascii=False)}")  # リクエストデータを出力
    response = requests.post(url, headers=headers, data=json.dumps(data))

    print(f"Received response with status code {response.status_code}")  # ステータスコードを出力

    if response.status_code == 200:
        if response_mode == "blocking":
            result = response.json()
            print(f"Response JSON: {result}")  # レスポンス全体を出力
            if "data" in result and "outputs" in result["data"] and "text" in result["data"]["outputs"]:
                return result["data"]["outputs"]["text"]
            else:
                print("Error: 'text' not found in the API response.")
        elif response_mode == "streaming":
            for chunk in response.iter_content(chunk_size=None):
                if chunk:
                    print(chunk.decode())
    else:
        print(f"Request failed with status code {response.status_code}")
        try:
            print(response.json())
        except json.JSONDecodeError:
            print(response.text)
```

- `run_workflow`関数は、Difyのワークフローを実行します。
- `url`: ワークフローを実行するためのAPIエンドポイント。
- `headers`: APIリクエストのヘッダー。認証キーとJSONコンテンツタイプが含まれます。
- `data`: リクエストデータ。`inputs`には入力テキスト、`response_mode`にはレスポンスモード、`user`にはユーザーIDが含まれます。
- `requests.post`: リクエストを送信します。
- レスポンスのステータスコードに応じて、適切な処理を行います。

### ワークフローの停止関数

```python
def stop_workflow(task_id, user):
    url = f"{BASE_URL}/workflows/{task_id}/stop"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    data = {"user": user}
    response = requests.post(url, headers=headers, data=json.dumps(data))

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Request failed with status code {response.status_code}")
```

- `stop_workflow`関数は、実行中のワークフローを停止します。
- `url`: ワークフローを停止するためのAPIエンドポイント。
- `data`: リクエストデータ。ユーザーIDが含まれます。
- リクエストを送信し、レスポンスを処理します。

### パラメータの取得関数

```python
def get_parameters(user):
    url = f"{BASE_URL}/parameters?user={user}"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Request failed with status code {response.status_code}")
```

- `get_parameters`関数は、APIからパラメータを取得します。
- `url`: パラメータを取得するためのAPIエンドポイント。
- リクエストを送信し、レスポンスを処理します。

### メイン処理

```python
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Run workflow")
    parser.add_argument("--input", type=str, required=True, help="Input file path")
    parser.add_argument("--response_mode", type=str, help="Response mode (blocking or streaming)")
    parser.add_argument("--user", type=str, default="example_user", help="User identifier")

    args = parser.parse_args()

    with open(args.input, 'r', encoding='utf-8') as file:
        input_text = file.read().strip()

    inputs = {"input_text": input_text}
    response_mode = args.response_mode if args.response_mode else "blocking"
    user = args.user

    result = run_workflow(inputs, response_mode, user)
    if result:
        print(result)
```

- コマンドライン引数を解析します。
  - `--input`: 入力ファイルのパス。
  - `--response_mode`: レスポンスモード（指定がない場合は"blocking"）。
  - `--user`: ユーザー識別子。
- 指定された入力ファイルからテキストを読み込みます。
- `run_workflow`関数を呼び出し、結果を表示します。

### 実行例

以下のコマンドでスクリプトを実行します：

```sh
python3 dify-call.py --input "/home/xxc/dify/DATA/input.txt" --user "user"
```
このスクリプトを使用することで、簡単にDify APIを利用してワークフローを実行できます。必要な情報を入力ファイルにまとめ、コマンドラインからスクリプトを実行するだけです。これにより、ワークフローの実行とレスポンスの取得が自動化され、効率的に作業を進めることができます。なお、レスポンスモードが指定されていない場合は、デフォルトで"blocking"になります。