task-board/example_agent.py

#!/usr/bin/env python3
"""
多智能体任务系统 - Python 客户端示例
"""

import requests
import time
import sys

class TaskAgent:
    def __init__(self, name, base_url="http://localhost:3000"):
        self.name = name
        self.base_url = base_url
        self.agent_id = None
        
    def register(self):
        """注册智能体"""
        try:
            response = requests.post(f"{self.base_url}/api/agents/register", json={
                "name": self.name,
                "type": "general",
                "capabilities": ["general_task", "data_processing"]
            })
            response.raise_for_status()
            data = response.json()
            self.agent_id = data["id"]
            print(f"✅ 智能体已注册: {self.agent_id}")
            return True
        except Exception as e:
            print(f"❌ 注册失败: {e}")
            return False
        
    def heartbeat(self, status="idle"):
        """发送心跳"""
        try:
            requests.post(
                f"{self.base_url}/api/agents/{self.agent_id}/heartbeat",
                json={"status": status},
                timeout=5
            )
        except Exception as e:
            print(f"⚠️  心跳发送失败: {e}")
        
    def get_pending_tasks(self):
        """获取待处理任务"""
        try:
            response = requests.get(
                f"{self.base_url}/api/tasks?status=pending",
                timeout=5
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"❌ 获取任务失败: {e}")
            return []
        
    def claim_task(self, task_id):
        """领取任务"""
        try:
            response = requests.post(
                f"{self.base_url}/api/tasks/{task_id}/claim",
                json={"agent_id": self.agent_id},
                timeout=5
            )
            response.raise_for_status()
            return True
        except Exception as e:
            print(f"⚠️  领取任务失败: {e}")
            return False
        
    def update_task(self, task_id, status, result=None):
        """更新任务状态"""
        try:
            payload = {"status": status}
            if result:
                payload["result"] = result
            response = requests.put(
                f"{self.base_url}/api/tasks/{task_id}",
                json=payload,
                timeout=5
            )
            response.raise_for_status()
        except Exception as e:
            print(f"❌ 更新任务失败: {e}")
        
    def execute_task(self, task):
        """执行任务（示例实现）"""
        print(f"🔄 正在执行任务: {task['title']}")
        print(f"   描述: {task.get('description', '无')}")
        print(f"   优先级: {task['priority']}")
        
        # 模拟任务执行
        time.sleep(3)
        
        return f"任务 '{task['title']}' 已由 {self.name} 成功完成"
        
    def run(self):
        """主循环"""
        if not self.register():
            print("❌ 无法注册智能体，退出")
            return
        
        print(f"🤖 智能体 '{self.name}' 开始运行...")
        print("按 Ctrl+C 退出\n")
        
        while True:
            try:
                # 发送心跳
                self.heartbeat("idle")
                
                # 获取待处理任务
                tasks = self.get_pending_tasks()
                
                if tasks:
                    task = tasks[0]
                    print(f"\n📋 发现新任务: {task['title']} (ID: {task['id'][:8]}...)")
                    
                    # 领取任务
                    if self.claim_task(task['id']):
                        print("✓ 任务已领取")
                        self.heartbeat("busy")
                        
                        # 执行任务
                        result = self.execute_task(task)
                        
                        # 更新任务状态
                        self.update_task(task['id'], "completed", result)
                        print(f"✅ 任务完成: {task['id'][:8]}...")
                        print(f"   结果: {result}\n")
                        
                        self.heartbeat("idle")
                    else:
                        print("⚠️  任务可能已被其他智能体领取\n")
                
                # 等待一段时间再检查
                time.sleep(5)
                
            except KeyboardInterrupt:
                print("\n\n👋 收到退出信号，智能体停止运行")
                break
            except Exception as e:
                print(f"❌ 运行时错误: {e}")
                time.sleep(5)

def main():
    if len(sys.argv) > 1:
        agent_name = sys.argv[1]
    else:
        agent_name = f"Python智能体-{int(time.time()) % 10000}"
    
    base_url = sys.argv[2] if len(sys.argv) > 2 else "http://localhost:3000"
    
    agent = TaskAgent(agent_name, base_url)
    agent.run()

if __name__ == "__main__":
    main()