myagents/.claude/agent/agent_worker.py

#!/usr/bin/env python3
"""Claude Agent Worker - 后台执行进程."""

import json
import os
import sys
import time
from pathlib import Path

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parents[3]))

try:
    from anthropic import Anthropic
except ImportError:
    print("Error: anthropic SDK not installed")
    sys.exit(1)

TASKS_FILE = Path(".claude/agent/tasks.jsonl")
RESULTS_DIR = Path(".claude/agent/results")
RESULTS_DIR.mkdir(parents=True, exist_ok=True)


def load_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 execute_task(task: dict):
    """使用 Claude API 执行任务."""
    client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

    task_id = task["id"]
    title = task["title"]
    description = task["description"]
    context = task.get("context", {})

    # 构建系统提示
    system_prompt = f"""You are an autonomous task execution agent.
Your job is to complete tasks by analyzing requirements and taking actions.

You have access to these tools via function calling:
- read_file: Read file contents
- write_file: Write to files
- edit_file: Edit existing files
- run_command: Execute shell commands
- list_files: List directory contents

Current workspace: {os.getcwd()}
Task ID: {task_id}

Be thorough and report your actions clearly."""

    # 构建用户消息
    user_message = f"""Task: {title}
Description: {description}
Context: {json.dumps(context, indent=2)}

Please complete this task step by step.
1. Analyze what needs to be done
2. Execute necessary actions
3. Report results

Start now."""

    print(f"[Agent] Processing task: {title}")

    try:
        # 调用 Claude API
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system_prompt,
            messages=[{"role": "user", "content": user_message}],
        )

        result = {
            "task_id": task_id,
            "status": "completed",
            "response": response.content[0].text if response.content else "",
            "completed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        }

        # 保存结果
        result_file = RESULTS_DIR / f"{task_id}.json"
        with open(result_file, "w") as f:
            json.dump(result, f, indent=2)

        print(f"[Agent] Task completed: {task_id}")
        return result

    except Exception as e:
        result = {
            "task_id": task_id,
            "status": "failed",
            "error": str(e),
            "completed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        }

        result_file = RESULTS_DIR / f"{task_id}.json"
        with open(result_file, "w") as f:
            json.dump(result, f, indent=2)

        print(f"[Agent] Task failed: {task_id}, error: {e}")
        return result


def update_task_status(task_id: str, status: str):
    """更新任务状态."""
    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")
                lines.append(json.dumps(task))
            except:
                lines.append(line.strip())

    with open(TASKS_FILE, "w") as f:
        f.write("\n".join(lines) + "\n")


def main():
    """主循环."""
    print("[Agent] Claude Agent Worker started")
    print(f"[Agent] Tasks file: {TASKS_FILE.absolute()}")
    print(f"[Agent] Results dir: {RESULTS_DIR.absolute()}")

    while True:
        try:
            tasks = load_tasks()

            for task in tasks:
                task_id = task["id"]
                update_task_status(task_id, "processing")
                execute_task(task)
                update_task_status(task_id, "completed")

            time.sleep(5)

        except KeyboardInterrupt:
            print("\n[Agent] Stopping...")
            break
        except Exception as e:
            print(f"[Agent] Error: {e}")
            time.sleep(5)


if __name__ == "__main__":
    main()