chatflow
import requests import json # API 配置 api_key = "app-xxx" # 替换为你的 API Key base_url = "http://192.168.1.100:8080/v1" endpoint = "/chat-messages" # 请求头 headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } # 请求体 data = { "inputs": {}, "query": "What are the specs of the iPhone 13 Pro Max?", "response_mode": "streaming", "conversation_id": "", "user": "ywxx" } # 发送请求 response = requests.post( url=f"{base_url}{endpoint}", headers=headers, data=json.dumps(data), stream=True # 启用流式响应 ) # 处理流式响应 if response.status_code == 200: for line in response.iter_lines(): if line: line = line.decode('utf-8') if line.startswith('data: '): try: data = json.loads(line[6:]) # 移除 'data: ' 前缀 print(data) except json.JSONDecodeError: continue else: print(f"Error: {response.status_code}") print(response.text)
工作流(workflows)
import requests # API 配置 api_key = "app-zzzz" base_url = "http://192.168.1.110:8080/v1" # 请求头 headers = { "Authorization": f"Bearer {api_key}" } # 发送请求 response = requests.get( url=f"{base_url}/workflows/logs", headers=headers ) # response = requests.get( # url=f"{base_url}/info", # headers=headers # ) # 检查响应 if response.status_code == 200: logs_data = response.json() print("工作流日志获取成功：") print(logs_data) else: print(f"请求失败：{response.status_code}") print(response.text)
