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

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)

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

インポートと定数の定義

import requests
import json
import argparse

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

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

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

ワークフローの実行関数

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: リクエストを送信します。
レスポンスのステータスコードに応じて、適切な処理を行います。

ワークフローの停止関数

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が含まれます。
リクエストを送信し、レスポンスを処理します。

パラメータの取得関数

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エンドポイント。
リクエストを送信し、レスポンスを処理します。

メイン処理

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関数を呼び出し、結果を表示します。

実行例

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

python3 dify-call.py --input "/home/xxc/dify/DATA/input.txt" --user "user"

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