---

title: 如何通过程序调用 OpenClaw Agent 实现自动化任务 date: 2026-03-22 tags: [OpenClaw, API, 自动化，Python] category: 开发教程

如何通过程序调用 OpenClaw Agent 实现自动化任务

💡 问题背景：如何通过代码调用 OpenClaw Agent，而不是只能通过 WebUI 对话？本文介绍多种编程方式调用 OpenClaw，实现调研、报告生成等自动化任务。

---

🎯 一、调用方式总览

方式	适用场景	难度
CLI 命令	脚本调用	⭐
Python SDK	深度集成	⭐⭐⭐
Gateway WebSocket	实时交互	⭐⭐⭐⭐
sessions_spawn 工具	子任务执行	⭐⭐

---

🔧 二、CLI 命令调用（推荐）

2.1 基础用法

bash
# 执行单个任务
openclaw agent --agent main --message "调研 AI 最新发展" --json

# 指定会话
openclaw agent --session-id abc123 --message "继续调研" --json

# 设置思考级别
openclaw agent --thinking high --message "生成架构设计" --json

2.2 Python 封装

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OpenClaw Agent CLI 封装
"""

import subprocess
import json
from typing import Dict, Any, Optional

class OpenClawAgent:
    """OpenClaw Agent 调用封装"""
    
    def __init__(self, agent_id: str = "main"):
        self.agent_id = agent_id
    
    def run(self, 
            message: str, 
            thinking: str = "medium",
            timeout: int = 600) -> Dict[str, Any]:
        """
        执行 Agent 任务
        
        Args:
            message: 任务描述
            thinking: 思考级别 (off/minimal/low/medium/high/xhigh)
            timeout: 超时时间（秒）
            
        Returns:
            执行结果
        """
        cmd = [
            "openclaw", "agent",
            "--agent", self.agent_id,
            "--message", message,
            "--thinking", thinking,
            "--timeout", str(timeout),
            "--json"
        ]
        
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout + 30
        )
        
        if result.returncode != 0:
            return {
                "success": False,
                "error": result.stderr
            }
        
        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            return {
                "success": True,
                "raw": result.stdout
            }
    
    def run_with_session(self,
                        session_id: str,
                        message: str,
                        **kwargs) -> Dict[str, Any]:
        """在指定会话中执行"""
        cmd = [
            "openclaw", "agent",
            "--agent", self.agent_id,
            "--session-id", session_id,
            "--message", message,
            "--json"
        ]
        # ... 同上
        
        return self._run_cmd(cmd)


# 使用示例
if __name__ == "__main__":
    agent = OpenClawAgent()
    
    # 执行调研任务
    result = agent.run("调研 2026 年 AI Agent 最新发展动态")
    
    if result.get("success"):
        print("✅ 调研完成")
        print(result.get("raw", result))
    else:
        print("❌ 调研失败:", result.get("error"))

2.3 实际案例：自动化调研

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动化调研报告生成
"""

from openclaw_agent import OpenClawAgent
import json
from datetime import datetime

class ResearchAgent:
    """研究 Agent"""
    
    def __init__(self):
        self.agent = OpenClawAgent()
        self.session_id = None
    
    def research(self, topic: str) -> dict:
        """
        执行完整调研流程
        """
        print(f"🔍 开始调研：{topic}")
        
        # 步骤 1：生成大纲
        print("\n📋 步骤 1: 生成调研大纲")
        outline_result = self.agent.run(
            f"为'{topic}'生成详细的调研大纲，列出主要方向"
        )
        
        # 步骤 2：分方向调研
        print("\n📊 步骤 2: 分方向深入调研")
        research_results = []
        
        # 这里可以解析大纲，然后对每个方向执行调研
        for direction in ["技术趋势", "市场动态", "主要玩家"]:
            result = self.agent.run(
                f"调研{topic}的{direction}，提供详细数据和案例"
            )
            research_results.append({
                "direction": direction,
                "result": result
            })
        
        # 步骤 3：生成报告
        print("\n📝 步骤 3: 生成完整报告")
        report_result = self.agent.run(
            f"根据前面的调研，生成'{topic}'的完整调研报告，包含：\n"
            "1. 执行摘要\n"
            "2. 技术趋势\n"
            "3. 市场分析\n"
            "4. 主要玩家\n"
            "5. 结论和建议"
        )
        
        return {
            "topic": topic,
            "timestamp": datetime.now().isoformat(),
            "outline": outline_result,
            "research": research_results,
            "report": report_result
        }


if __name__ == "__main__":
    researcher = ResearchAgent()
    report = researcher.research("AI Agent 框架")
    
    # 保存报告
    with open(f"report_{datetime.now().strftime('%Y%m%d')}.json", "w") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    
    print("\n✅ 报告已保存！")

---

🌐 三、Gateway WebSocket 调用

3.1 连接 Gateway

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
通过 WebSocket 连接 OpenClaw Gateway
"""

import asyncio
import json
import websockets

class OpenClawGateway:
    """OpenClaw Gateway WebSocket 客户端"""
    
    def __init__(self, 
                 url: str = "ws://localhost:18789",
                 token: str = None):
        self.url = url
        self.token = token
        self.ws = None
        self.message_id = 0
    
    async def connect(self):
        """连接到 Gateway"""
        self.ws = await websockets.connect(self.url)
        
        # 发送连接请求
        connect_msg = {
            "type": "req",
            "id": self._gen_id(),
            "method": "connect",
            "params": {
                "client": {
                    "id": "python-client",
                    "version": "1.0.0"
                },
                "role": "operator",
                "scopes": ["operator.read", "operator.write"],
                "auth": {"token": self.token} if self.token else {}
            }
        }
        
        await self.ws.send(json.dumps(connect_msg))
        response = await self.ws.recv()
        print("连接成功:", json.loads(response))
    
    async def send_message(self, 
                          message: str,
                          agent_id: str = "main") -> dict:
        """发送消息给 Agent"""
        msg = {
            "type": "req",
            "id": self._gen_id(),
            "method": "agent.send",
            "params": {
                "agent": agent_id,
                "message": message
            }
        }
        
        await self.ws.send(json.dumps(msg))
        response = await self.ws.recv()
        return json.loads(response)
    
    def _gen_id(self) -> str:
        self.message_id += 1
        return f"msg-{self.message_id}"
    
    async def close(self):
        if self.ws:
            await self.ws.close()


# 使用示例
async def main():
    gateway = OpenClawGateway(token="your-token")
    await gateway.connect()
    
    # 执行任务
    result = await gateway.send_message(
        "调研最新 AI 技术趋势"
    )
    
    print("执行结果:", result)
    
    await gateway.close()

if __name__ == "__main__":
    asyncio.run(main())

---

🛠️ 四、sessions_spawn 工具调用

4.1 创建子 Agent

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
使用 sessions_spawn 创建子 Agent 执行任务
"""

import subprocess
import json

def spawn_agent_task(task: str, 
                    agent_id: str = "main",
                    mode: str = "run") -> dict:
    """
    创建子 Agent 任务
    
    Args:
        task: 任务描述
        agent_id: Agent ID
        mode: run（一次性）或 session（持久会话）
    """
    cmd = [
        "openclaw",
        "sessions", "spawn",
        "--task", task,
        "--mode", mode,
        "--agent", agent_id,
        "--json"
    ]
    
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True
    )
    
    return json.loads(result.stdout)


# 使用示例
if __name__ == "__main__":
    # 创建调研任务
    result = spawn_agent_task(
        "调研 2026 年 AI Agent 发展趋势，生成详细报告",
        agent_id="main",
        mode="run"
    )
    
    print("任务创建成功:", result)
    print("任务 ID:", result.get("sessionKey"))

4.2 查询任务状态

python
def check_task_status(session_key: str) -> dict:
    """查询任务状态"""
    cmd = [
        "openclaw",
        "sessions", "history",
        "--session-key", session_key,
        "--json"
    ]
    
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True
    )
    
    return json.loads(result.stdout)


# 轮询任务状态
import time

def wait_for_completion(session_key: str, 
                       timeout: int = 1800) -> dict:
    """等待任务完成"""
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        status = check_task_status(session_key)
        
        if status.get("completed"):
            return status
        
        print(f"任务进行中... {time.time() - start_time:.0f}s")
        time.sleep(10)
    
    return {"error": "超时"}

---

📊 五、完整案例：自动化报告生成系统

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动化报告生成系统
通过 OpenClaw Agent 实现调研、分析、报告生成全流程
"""

import json
import time
from datetime import datetime
from pathlib import Path
from openclaw_agent import OpenClawAgent

class ReportGenerator:
    """报告生成器"""
    
    def __init__(self, output_dir: str = "./reports"):
        self.agent = OpenClawAgent()
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
    
    def generate_report(self, 
                       topic: str,
                       sections: list = None) -> dict:
        """
        生成完整报告
        
        Args:
            topic: 报告主题
            sections: 章节列表（可选，默认自动生成）
        """
        print(f"🚀 开始生成报告：{topic}")
        
        report = {
            "topic": topic,
            "created_at": datetime.now().isoformat(),
            "sections": []
        }
        
        # 步骤 1：生成大纲
        print("\n📋 生成大纲...")
        outline = self._generate_outline(topic)
        report["outline"] = outline
        
        # 步骤 2：分章节撰写
        sections = sections or outline.get("sections", [])
        for i, section in enumerate(sections, 1):
            print(f"\n📝 撰写章节 {i}/{len(sections)}: {section}")
            content = self._write_section(topic, section)
            report["sections"].append({
                "title": section,
                "content": content
            })
        
        # 步骤 3：生成摘要和结论
        print("\n📊 生成摘要和结论...")
        summary = self._generate_summary(report)
        report["summary"] = summary
        
        # 保存报告
        self._save_report(report)
        
        return report
    
    def _generate_outline(self, topic: str) -> dict:
        """生成大纲"""
        result = self.agent.run(
            f"为'{topic}'生成详细的报告大纲，返回 JSON 格式：\n"
            "{{\"sections\": [\"章节 1\", \"章节 2\", ...]}}"
        )
        return result
    
    def _write_section(self, topic: str, section: str) -> str:
        """撰写章节"""
        result = self.agent.run(
            f"撰写'{topic}'报告中'{section}'章节的内容，要求：\n"
            "1. 详细、专业\n"
            "2. 包含数据和案例\n"
            "3. 至少 500 字"
        )
        return result.get("raw", "")
    
    def _generate_summary(self, report: dict) -> str:
        """生成摘要"""
        result = self.agent.run(
            f"根据以下报告内容，生成 300 字的执行摘要和结论：\n"
            f"主题：{report['topic']}\n"
            f"章节：{len(report['sections'])}个"
        )
        return result.get("raw", "")
    
    def _save_report(self, report: dict):
        """保存报告"""
        filename = f"{report['topic'][:20]}_{datetime.now().strftime('%Y%m%d')}.json"
        filepath = self.output_dir / filename
        
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        print(f"\n✅ 报告已保存：{filepath}")


if __name__ == "__main__":
    generator = ReportGenerator()
    
    report = generator.generate_report(
        "2026 年 AI Agent 发展趋势",
        sections=[
            "技术趋势",
            "市场分析",
            "主要玩家",
            "应用场景",
            "挑战与机遇"
        ]
    )
    
    print("\n🎉 报告生成完成！")

---

🎯 六、最佳实践

6.1 错误处理

python
def safe_agent_call(message: str, retries: int = 3) -> dict:
    """带重试的安全调用"""
    for i in range(retries):
        try:
            result = agent.run(message)
            if result.get("success"):
                return result
        except Exception as e:
            print(f"第{i+1}次失败：{e}")
            time.sleep(2 ** i)  # 指数退避
    
    return {"success": False, "error": "多次重试失败"}

6.2 进度追踪

python
def track_progress(task_id: str):
    """追踪任务进度"""
    while True:
        status = check_task_status(task_id)
        print(f"进度：{status.get('progress', 0)}%")
        
        if status.get("completed"):
            break
        
        time.sleep(5)

6.3 成本控制

python
# 设置 token 预算
def budget_aware_call(message: str, max_tokens: int = 10000):
    """带 token 预算的调用"""
    result = agent.run(
        message,
        extra_args={"maxTokens": max_tokens}
    )
    print(f"消耗 token: {result.get('usage', {})}")
    return result

---

📚 七、参考资料

• OpenClaw CLI 文档
• Gateway 协议
• Agent 配置
• own-code-dashboard 项目

---

本文示例代码：~/Documents/own-code-dashboard/scripts/openclaw-examples/