Coverage for agentos/orchestration/a2a_router.py: 67%

1""" 

2A2A协议路由 — 跨框架Agent互操作。 

3基因来源: Google ADK A2A Protocol 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from enum import StrEnum 

10from typing import Any 

11 

12 

13class TaskStatus(StrEnum): 

14 """任务状态枚举。""" 

15 

16 PENDING = "pending" 

17 IN_PROGRESS = "in_progress" 

18 COMPLETED = "completed" 

19 FAILED = "failed" 

20 

21 

22@dataclass 

23class AgentCard: 

24 """Agent名片 — A2A协议中Agent互相发现的基础。""" 

25 

26 id: str 

27 name: str 

28 description: str 

29 capabilities: list[str] = field(default_factory=list) 

30 endpoint: str = "" 

31 protocol_version: str = "1.0" 

32 

33 

34@dataclass 

35class Task: 

36 """结构化任务 — A2A协议的任务定义。""" 

37 

38 id: str 

39 description: str 

40 input_data: dict[str, Any] = field(default_factory=dict) 

41 status: TaskStatus = TaskStatus.PENDING 

42 

43 

44@dataclass 

45class TaskResult: 

46 """Result of an A2A routed task execution.""" 

47 

48 task_id: str 

49 output: str 

50 artifacts: list[str] = field(default_factory=list) 

51 error: str | None = None 

52 

53 

54class A2ARouter: 

55 """ 

56 A2A协议路由 — 让不同框架构建的Agent相互通信。 

57 

58 核心流程: 

59 1. Agent Card 注册 → 互相发现 

60 2. Task 委派 → 结构化任务传递 

61 3. Message 协商 → 多轮异步通信 

62 4. Artifact 返回 → 产物传递 

63 """ 

64 

65 def __init__(self): 

66 self.local_agents: dict[str, AgentCard] = {} 

67 self.remote_agents: dict[str, AgentCard] = {} 

68 self._task_results: dict[str, TaskResult] = {} 

69 

70 def register(self, card: AgentCard) -> None: 

71 """Register an agent card (compliance test entry point).""" 

72 self.local_agents[card.id] = card 

73 

74 def register_local(self, card: AgentCard): 

75 self.local_agents[card.id] = card 

76 

77 def discover_remote(self, cards: list[AgentCard]): 

78 for card in cards: 

79 self.remote_agents[card.id] = card 

80 

81 def find_agent(self, capability: str) -> AgentCard | None: 

82 """按能力查找Agent。""" 

83 all_agents = {**self.local_agents, **self.remote_agents} 

84 for agent in all_agents.values(): 

85 if capability.lower() in [c.lower() for c in agent.capabilities]: 

86 return agent 

87 return None 

88 

89 def delegate(self, task: Task, agent_id: str | None = None) -> TaskResult: 

90 """ 

91 任务委派。实际生产环境中会通过A2A协议异步调用远程Agent。 

92 当前为本地模拟实现。 

93 """ 

94 task.status = TaskStatus.IN_PROGRESS 

95 

96 # 模拟异步执行 

97 result = TaskResult( 

98 task_id=task.id, 

99 output=f"Agent {agent_id or 'unknown'} processed: {task.description}", 

100 ) 

101 task.status = TaskStatus.COMPLETED 

102 self._task_results[task.id] = result 

103 return result 

104 

105 def list_agents(self) -> list[AgentCard]: 

106 return list(self.local_agents.values()) + list(self.remote_agents.values())