Zhengshou Lai
b98d35c387
- Click entrypoint: mybot update (make install / pip -e), mybot chat (claude in repo) - Optional --dangerously-skip-permissions for Claude Code - Makefile install; hatchling package layout - .gitignore: venv, build artifacts, workspace (local symlinks) Made-with: Cursor
150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Claude Agent Worker - 使用 Claude CLI 执行智能任务."""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
TASKS_FILE = Path(".claude/agent/tasks.jsonl")
|
|
RESULTS_DIR = Path(".claude/agent/results")
|
|
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def load_pending_tasks():
|
|
"""加载待处理任务."""
|
|
if not TASKS_FILE.exists():
|
|
return []
|
|
|
|
tasks = []
|
|
with open(TASKS_FILE) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
task = json.loads(line)
|
|
if task.get("status") == "pending":
|
|
tasks.append(task)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return tasks
|
|
|
|
|
|
def update_task_status(task_id: str, status: str, result: dict | None = None):
|
|
"""更新任务状态."""
|
|
if not TASKS_FILE.exists():
|
|
return
|
|
|
|
lines = []
|
|
with open(TASKS_FILE) as f:
|
|
for line in f:
|
|
try:
|
|
task = json.loads(line)
|
|
if task.get("id") == task_id:
|
|
task["status"] = status
|
|
if status == "processing":
|
|
task["started_at"] = time.strftime("%Y-%m-%dT%H:%M:%S")
|
|
if result:
|
|
task["result"] = result
|
|
lines.append(json.dumps(task, ensure_ascii=False))
|
|
except:
|
|
lines.append(line.strip())
|
|
|
|
with open(TASKS_FILE, "w") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
|
|
|
|
def execute_with_claude(task: dict) -> dict:
|
|
"""使用 Claude CLI 执行任务."""
|
|
task_id = task["id"]
|
|
title = task["title"]
|
|
description = task["description"]
|
|
|
|
# 构建提示词
|
|
prompt = f"""You are an autonomous task execution agent.
|
|
Complete the following task autonomously.
|
|
|
|
TASK: {title}
|
|
DESCRIPTION: {description}
|
|
|
|
INSTRUCTIONS:
|
|
1. Analyze what needs to be done
|
|
2. Use tools to complete the task (read files, run commands, etc.)
|
|
3. Save a summary of your actions and results
|
|
4. Be concise but thorough
|
|
|
|
Start executing now. Report your progress and final result."""
|
|
|
|
print(f"[Worker] Executing: {title}")
|
|
|
|
try:
|
|
# 调用 Claude CLI
|
|
result = subprocess.run(
|
|
["claude", "-p", prompt],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300, # 5分钟超时
|
|
cwd=Path.cwd(),
|
|
)
|
|
|
|
output = result.stdout if result.returncode == 0 else f"Error: {result.stderr}"
|
|
|
|
return {
|
|
"success": result.returncode == 0,
|
|
"output": output[:5000], # 限制长度
|
|
"returncode": result.returncode,
|
|
}
|
|
|
|
except subprocess.TimeoutExpired:
|
|
return {"success": False, "output": "Task timed out", "error": "timeout"}
|
|
except Exception as e:
|
|
return {"success": False, "output": "", "error": str(e)}
|
|
|
|
|
|
def main():
|
|
"""主循环."""
|
|
print("[Worker] Claude Agent Worker started")
|
|
print(f"[Worker] Tasks: {TASKS_FILE.absolute()}")
|
|
print(f"[Worker] Results: {RESULTS_DIR.absolute()}")
|
|
|
|
while True:
|
|
try:
|
|
tasks = load_pending_tasks()
|
|
|
|
for task in tasks:
|
|
task_id = task["id"]
|
|
update_task_status(task_id, "processing")
|
|
|
|
result = execute_with_claude(task)
|
|
|
|
# 保存结果
|
|
result_file = RESULTS_DIR / f"{task_id}.json"
|
|
with open(result_file, "w") as f:
|
|
json.dump(
|
|
{
|
|
"task_id": task_id,
|
|
"result": result,
|
|
"completed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
},
|
|
f,
|
|
indent=2,
|
|
)
|
|
|
|
update_task_status(task_id, "completed", result)
|
|
print(f"[Worker] Completed: {task_id}")
|
|
|
|
time.sleep(5)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n[Worker] Stopping...")
|
|
break
|
|
except Exception as e:
|
|
print(f"[Worker] Error: {e}")
|
|
time.sleep(5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|