在 Coze 插件开发中，将 Postman 调试通过的接口转换为 Python 代码是一个常见需求。下面我将详细解析这个转换过程，并提供完整的代码示例和实践指导。 ## 1. Postman 接口信息分析 首先需要从 Postman 中提取完整的接口信息： | 信息类型 | 说明 | 示例 | |---------|------|------| | 请求方法 | HTTP 方法类型 | GET、POST、PUT 等 | | 请求URL | 完整的接口地址 | `https://api.coze.com/v1/chat` | | 请求头 | 认证和内容类型头 | `Authorization: Bearer xxx` | | 请求体 | 发送的数据内容 | JSON 格式参数 | | 查询参数 | URL 参数 | `?user_id=123&session_id=abc` | ## 2. Python 请求库选择 根据不同的使用场景，可以选择以下 Python 库： | 库名称 | 适用场景 | 特点 | |--------|----------|------| | requests | 通用 HTTP 请求 | 简单易用，功能全面 | | aiohttp | 异步高性能请求 | 适合高并发场景 | | httpx | 现代化 HTTP 客户端 | 同步异步都支持 | ## 3. 完整转换示例 ### 3.1 基础 GET 请求转换 **Postman 配置：** - 方法：GET - URL：`https://api.coze.com/v1/bots` - 头部：`Authorization: Bearer your_api_key` **对应 Python 代码：** ```python import requests def get_bots_list(api_key): """ 获取 Coze 机器人列表 :param api_key: Coze API 密钥 :return: 响应数据 """ url = "https://api.coze.com/v1/bots" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } try: response = requests.get(url, headers=headers) response.raise_for_status() # 检查 HTTP 错误 return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None # 使用示例 api_key = "your_coze_api_key_here" bots = get_bots_list(api_key) if bots: print(f"获取到 {len(bots.get('data', []))} 个机器人") ``` ### 3.2 复杂 POST 请求转换 **Postman 配置：** - 方法：POST - URL：`https://api.coze.com/v1/chat` - 头部：`Authorization: Bearer your_api_key` - Body (raw JSON): ```json { "bot_id": "123456", "user_id": "user_001", "query": "你好，今天天气怎么样？", "stream": false } ``` **对应 Python 代码：** ```python import requests import json def send_chat_message(api_key, bot_id, user_id, query): """ 向 Coze 机器人发送聊天消息 :param api_key: API 密钥 :param bot_id: 机器人ID :param user_id: 用户ID :param query: 用户查询 :return: 聊天响应 """ url = "https://api.coze.com/v1/chat" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "bot_id": bot_id, "user_id": user_id, "query": query, "stream": False } try: response = requests.post(url, headers=headers, json=payload) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: print(f"HTTP 错误: {e}") if response.status_code == 400: print("请求参数错误") elif response.status_code == 401: print("认证失败") elif response.status_code == 404: print("接口不存在") elif response.status_code == 500: print("服务器内部错误") return None except requests.exceptions.RequestException as e: print(f"网络请求异常: {e}") return None # 使用示例 api_key = "your_api_key" bot_id = "123456" user_id = "user_001" query = "你好，今天天气怎么样？" result = send_chat_message(api_key, bot_id, user_id, query) if result: print(f"机器人回复: {result.get('content', '')}") ``` ## 4. 高级功能实现 ### 4.1 支持流式响应 对于需要实时响应的场景，可以处理流式数据： ```python import requests import json def stream_chat_message(api_key, bot_id, user_id, query): """ 流式聊天消息处理 """ url = "https://api.coze.com/v1/chat" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "bot_id": bot_id, "user_id": user_id, "query": query, "stream": True } try: response = requests.post(url, headers=headers, json=payload, stream=True) response.raise_for_status() for line in response.iter_lines(): if line: line_data = line.decode('utf-8') if line_data.startswith('data: '): try: json_data = json.loads(line_data[6:]) if 'content' in json_data: yield json_data['content'] except json.JSONDecodeError: continue except Exception as e: print(f"流式请求失败: {e}") # 使用示例 for content in stream_chat_message(api_key, bot_id, user_id, query): print(content, end='', flush=True) ``` ### 4.2 异步请求实现 对于高并发场景，使用 aiohttp 实现异步请求： ```python import aiohttp import asyncio async def async_chat_message(api_key, bot_id, user_id, query): """ 异步发送聊天消息 """ url = "https://api.coze.com/v1/chat" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "bot_id": bot_id, "user_id": user_id, "query": query, "stream": False } async with aiohttp.ClientSession() as session: try: async with session.post(url, headers=headers, json=payload) as response: if response.status == 200: return await response.json() else: print(f"请求失败，状态码: {response.status}") return None except aiohttp.ClientError