knowledge/coze/coze_hook.py

import requests
import json
import os

class CozeHook(object):
    def __init__(self):
        self.url = "https://api.coze.cn/v1/workflow/run"
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": "Bearer " + os.getenv('COZE_API_KEY')
        }
        self.hook_id = os.getenv('COZE_BOT_ID')
    def call_coze_api(self, image_url_list, hook_id=None):
        url = self.url
        headers = self.headers
        payload = {
            "workflow_id": hook_id or self.hook_id,
            "parameters": {
                "image_url_list": image_url_list
            }
        }
        response = requests.post(url, json=payload, headers=headers, timeout=600)
        response.raise_for_status()
        return response.json()
    
    def run(self, image_url_list): 
        resp = self.call_coze_api(image_url_list)
        return resp