为了通过 Python 代码连接 OpenClaw 并提交用户需求让其处理，核心在于调用 OpenClaw 提供的 API 接口，将自然语言需求发送给其 AI Agent 工作流，并接收处理结果。根据参考资料，OpenClaw 可以作为本地化的 AI 网关或调度中心，支持通过 RESTful API 被外部程序调用[ref_6]。以下是具体的实现方法和步骤。 ### 1. OpenClaw API 基础 OpenClaw 的核心是一个 AI Agent 调度框架，其运行时（如 `Agent Runner` 或 `Gateway`）会暴露 HTTP 端点。开发者的主要操作就是向这些端点发送包含用户需求的请求。关键概念如下表所示： | 概念 | 说明 | 参考 | | :--- | :--- | :--- | | **Gateway** | 核心网关组件，负责接收外部请求、路由给具体的 Agent 或 Skill 执行。通常运行在特定端口（如 `8080`） | [ref_2] | | **Agent** | 执行具体任务的智能体，例如需求分析师、开发工程师等，可以通过 API 触发其工作流 | [ref_4] | | **Skill** | 可复用的功能模块（如代码生成、进程监控），Agent 可以调用 Skill 来完成复杂任务 | [ref_3][ref_5] | | **会话** | 为保持上下文连贯性，建议使用会话 ID (`session_id`) 来关联同一用户的多次交互 | [ref_6] | ### 2. Python 连接与调用方法 实现连接主要分为三步：**环境准备**、**构建请求**、**发送并处理响应**。 #### 步骤一：环境准备与依赖安装 确保 Python 环境已安装 `requests` 库，用于发送 HTTP 请求。 ```python # 安装 requests 库 (如果尚未安装) # pip install requests ``` #### 步骤二：确定 OpenClaw 服务端点 需要知道 OpenClaw 服务的 **基础 URL** 和 **具体的 API 路径**。通常，本地部署的 OpenClaw Gateway 地址类似 `http://localhost:8080`。具体的 API 路径可能因部署方式和版本而异，常见的有： * 触发一个指定 Agent 的执行：`POST /api/v1/agents/{agent_name}/execute` * 调用一个通用任务端点：`POST /api/v1/task/execute` * 直接调用 Skill：`POST /api/v1/skills/{skill_name}/execute` 具体路径请参考您的 OpenClaw 部署文档。假设我们使用一个通用的任务执行端点。 #### 步骤三：编写 Python 客户端代码 以下是一个完整的 Python 示例，演示如何连接 OpenClaw 并提交用户需求。 ```python import requests import json import uuid class OpenClawClient: """ OpenClaw API Python 客户端 用于提交用户需求并获取AI思考与执行结果。 """ def __init__(self, base_url="http://localhost:8080", api_key=None): """ 初始化客户端 :param base_url: OpenClaw Gateway 的基础地址 :param api_key: 可选，用于身份验证的API密钥 """ self.base_url = base_url.rstrip('/') self.headers = { 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'application/json' } if api_key: self.headers['Authorization'] = f'Bearer {api_key}' # 生成一个会话ID，用于关联同一用户的多次交互[ref_6] self.session_id = str(uuid.uuid4()) def submit_requirement(self, user_input, agent_name=None, skill_name=None): """ 向OpenClaw提交用户需求 :param user_input: 用户的需求描述，例如“开发一个简单的Flask API，返回当前时间” :param agent_name: 可选，指定由哪个Agent来处理（例如'developer'） :param skill_name: 可选，指定调用哪个Skill :return: OpenClaw的响应结果 """ # 构建请求体，核心是传递用户输入和会话上下文[ref_6] payload = { "session_id": self.session_id, # 使用固定的会话ID保持对话连续性 "user_input": user_input, "parameters": {} # 可根据需要传递额外参数 } # 确定请求的API端点 if agent_name: endpoint = f"{self.base_url}/api/v1/agents/{agent_name}/execute" elif skill_name: endpoint = f"{self.base_url}/api/v1/skills/{skill_name}/execute" else: # 默认为通用任务端点（假设存在） endpoint = f"{self.base_url}/api/v1/task/execute" try: # 发送POST请求[ref_6] response = requests.post( url=endpoint, headers=self.headers, data=json.dumps(payload, ensure_ascii=False).encode('utf-8'), timeout=120 # 任务可能耗时较长，设置较长超时[ref_6] ) response.raise_for_status() # 如果状态码不是200，抛出异常 result = response.json() return result except requests.exceptions.ConnectionError: return {"error": f"无法连接到OpenClaw服务，请检查服务是否运行在 {self.base_url}"} except requests.exceptions.Timeout: return {"error": "请求超时，OpenClaw处理时间过长"} except requests.exceptions.RequestException as e: return {"error": f"请求发生错误: {str(e)}"} except json.JSONDecodeError: return {"error": "OpenClaw返回了非JSON格式的响应"} def get_thought_process(self, task_id): """ 如果OpenClaw支持异步任务和进度查询，可以通过此方法获取任务的思考过程或中间状态。 :param task_id: 提交任务后返回的任务ID :return: 任务状态或思考日志 """ # 假设存在一个查询任务状态的端点 endpoint = f"{self.base_url}/api/v1/tasks/{task_id}/status" try: response = requests.get(endpoint, headers=self.headers, timeout=30) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": f"查询任务状态失败: {str(e)}"} # 使用示例 if __name__ == "__main__": # 1. 初始化客户端，连接到本地OpenClaw服务 client = OpenClawClient(base_url="http://localhost:8080") # 2. 定义用户需求 requirement = """ 请开发一个Python函数，名为`calculate_stats`，接收一个数字列表作为输入。 函数需要计算并返回这个列表的均值、最大值和最小值。 请生成完整的代码，并包含一个使用示例。 """ # 3. 提交需求，让OpenClaw的AI Agent（如开发工程师）进行思考和执行 print("正在向OpenClaw提交需求...") result = client.submit_requirement( user_input=requirement, agent_name="developer" # 假设存在名为'developer'的Agent[ref_4] ) # 4. 处理并打印结果 print("\n=== OpenClaw 响应 ===") if "error" in result: print(f"错误: {result['error']}") else: # 响应结构可能包含思考过程、生成的代码、执行结果等[ref_1][ref_4] # 这里根据实际响应格式进行解析 print(json.dumps(result, indent=2, ensure_ascii=False)) # 例如，如果返回了生成的代码 if "generated_code" in result: print("\n--- 生成的代码 ---") print(result["generated_code"]) if "execution_result" in result: print("\n--- 执行结果 ---") print(result["execution_result"]) ``` ### 3. 关键实现细节与最佳实践 在编写连接代码时，需要注意以下几点以确保稳定性和可用性： 1. **会话管理 (`session_id`)**: 为每个用户或对话线程分配一个唯一的 `session_id` 并贯穿多次请求，这有助于 OpenClaw 维护上下文，进行连贯的“思考”[ref_6]。 2. **超时与重试**: AI 任务处理时间不确定，代码中设置了较长的超时（如120秒）。在生产环境中，应考虑实现异步调用（提交任务后轮询状态）并加入重试逻辑[ref_6]。 3. **错误处理**: 必须妥善处理网络错误、服务不可用、响应格式异常等情况，如上例中的 `try-except` 块所示。 4. **指定处理单元**: 通过 `agent_name` 或 `skill_name` 参数，可以更精确地指定由 OpenClaw 中的哪个专业化模块来处理需求。例如，将代码生成需求发送给 `developer` Agent，或将服务器监控需求发送给 `process_monitor` Skill[ref_3][ref_4]。 5. **企业级集成考量**: 在 Java/Spring Boot 等企业后端集成 OpenClaw 时，推荐的模式是封装一个 HTTP 客户端，并加入连接池、熔断器（如 Hystrix/Sentinel）、审计日志等生产级特性。虽然这里是 Python 示例，但原理相通，`requests.Session` 可以用于连接复用[ref_6]。 ### 4. 应用场景示例 利用上述 Python 连接代码，可以实现多种自动化场景： | 场景 | 用户需求示例 | 预期的 OpenClaw 动作 | | :--- | :--- | :--- | | **自动代码生成** | “创建一个Flask应用，提供 `/weather` 接口，返回模拟的天气数据。” | `developer` Agent 被触发，分析需求后调用代码生成 Skill，最终返回完整的 Python 文件[ref_1]。 | | **运维自动化** | “检查服务器上进程 `ai_chat.py` 是否在运行，如果挂了就重启它。” | `system_agent` 调用 `agent-bootup` 这个进程监控 Skill 执行检查与重启逻辑[ref_3]。 | | **多Agent协作研发** | “我需要一个用户登录模块，包含数据库设计和API。” | 需求被提交后，OpenClaw 内部的 `Director` 会协调 `Requirement Analyst`、`Developer`、`Code Reviewer` 等多个 Sub-Agent 协作完成分析、编码和审查，最终返回完整方案[ref_4]。 | 总结来说，通过 Python 代码连接 OpenClaw 的核心是**构造符合其 API 规范的 HTTP 请求**，将自然语言需求、必要的参数以及会话标识发送至正确的端点。OpenClaw 的后端 AI 工作流（由多个 Agent 和 Skill 构成）会接手处理这些需求，完成从“思考”到“执行”的全过程，并将结果通过 API 响应返回给调用方[ref_1][ref_4][ref_6]。