Coverage for agentos/protocols/agent_card.py: 42%

1""" 

2AgentOS v1.2.0 — Agent Card 服务发现协议。 

3 

4基因来源: Google A2A (Agent-to-Agent) Agent Card 规范 

5 

6Agent Card 是标准化的 Agent 自描述卡片，支持: 

7- 发布/发现: Agent 发布自身能力，其他 Agent 按需发现 

8- 能力匹配: 按 domain / capability / keyword 搜索匹配 

9- 本地+远程: 文件系统本地发现 + HTTP 端点远程发现 

10- JSON 序列化: 完整的 export/import 往返，兼容 A2A 生态 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16from dataclasses import dataclass, field, asdict 

17from typing import Any, Dict, List, Optional 

18 

19 

20# ── AgentCard ─────────────────────────────────── 

21 

22@dataclass 

23class AgentCard: 

24 """Agent 自描述卡片，A2A 兼容。 

25 

26 使用方式: 

27 card = AgentCard( 

28 name="data-analyzer", 

29 description="数据分析Agent，支持SQL/Pandas/可视化", 

30 version="1.0.0", 

31 url="http://localhost:8000/agent", 

32 capabilities=["analysis", "coding"], 

33 skills=["sql-query", "pandas-transform", "chart-generate"], 

34 input_schema={"type": "object", "properties": {"query": {"type": "string"}}}, 

35 output_schema={"type": "object", "properties": {"result": {"type": "string"}}}, 

36 ) 

37 """ 

38 

39 name: str 

40 description: str 

41 version: str 

42 url: str = "" 

43 capabilities: List[str] = field(default_factory=list) 

44 skills: List[str] = field(default_factory=list) 

45 input_schema: Dict[str, Any] = field(default_factory=dict) 

46 output_schema: Dict[str, Any] = field(default_factory=dict) 

47 provider: str = "" 

48 metadata: Dict[str, Any] = field(default_factory=dict) 

49 tags: List[str] = field(default_factory=list) 

50 

51 def to_dict(self) -> Dict[str, Any]: 

52 """导出为字典，保留所有字段（含空值）。""" 

53 return asdict(self) 

54 

55 def to_json(self, indent: int = 2) -> str: 

56 """导出为 JSON 字符串。""" 

57 return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent) 

58 

59 @classmethod 

60 def from_dict(cls, data: Dict[str, Any]) -> "AgentCard": 

61 """从字典重建 AgentCard。""" 

62 return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) 

63 

64 @classmethod 

65 def from_json(cls, json_str: str) -> "AgentCard": 

66 """从 JSON 字符串重建。""" 

67 return cls.from_dict(json.loads(json_str)) 

68 

69 def matches_query(self, query: str) -> bool: 

70 """模糊匹配：检查 query 是否命中 name/description/skills/tags。""" 

71 q = query.lower() 

72 if q in self.name.lower(): 

73 return True 

74 if q in self.description.lower(): 

75 return True 

76 for skill in self.skills: 

77 if q in skill.lower(): 

78 return True 

79 for tag in self.tags: 

80 if q in tag.lower(): 

81 return True 

82 return False 

83 

84 def has_capability(self, capability: str) -> bool: 

85 return capability in self.capabilities 

86 

87 def has_skill(self, skill: str) -> bool: 

88 return skill in self.skills 

89 

90 def has_tag(self, tag: str) -> bool: 

91 return tag in self.tags 

92 

93 

94# ── AgentCardRegistry ─────────────────────────── 

95 

96@dataclass 

97class AgentCardRegistry: 

98 """Agent Card 注册中心。 

99 

100 支持注册、注销、搜索、过滤。 

101 """ 

102 

103 cards: Dict[str, AgentCard] = field(default_factory=dict) 

104 

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

106 """注册一张 Agent Card（同名覆盖）。""" 

107 self.cards[card.name] = card 

108 

109 def unregister(self, name: str) -> Optional[AgentCard]: 

110 """注销并返回被移除的卡片，不存在返回 None。""" 

111 return self.cards.pop(name, None) 

112 

113 def get(self, name: str) -> Optional[AgentCard]: 

114 """按名称查找。""" 

115 return self.cards.get(name) 

116 

117 def list_all(self) -> List[AgentCard]: 

118 """列出所有注册的卡片。""" 

119 return list(self.cards.values()) 

120 

121 def find_by_query(self, query: str) -> List[AgentCard]: 

122 """按关键词搜索（匹配 name/description/skills/tags）。""" 

123 return [c for c in self.cards.values() if c.matches_query(query)] 

124 

125 def find_by_capability(self, capability: str) -> List[AgentCard]: 

126 """按能力关键词查找。""" 

127 return [c for c in self.cards.values() if c.has_capability(capability)] 

128 

129 def find_by_skill(self, skill: str) -> List[AgentCard]: 

130 """按技能关键词查找。""" 

131 return [c for c in self.cards.values() if c.has_skill(skill)] 

132 

133 def find_by_tag(self, tag: str) -> List[AgentCard]: 

134 """按标签查找。""" 

135 return [c for c in self.cards.values() if c.has_tag(tag)] 

136 

137 def export_all(self, filepath: str) -> None: 

138 """将所有卡片导出到 JSON 文件。""" 

139 data = {name: card.to_dict() for name, card in self.cards.items()} 

140 with open(filepath, "w", encoding="utf-8") as f: 

141 json.dump(data, f, ensure_ascii=False, indent=2) 

142 

143 def import_from_file(self, filepath: str) -> int: 

144 """从 JSON 文件导入卡片到注册中心，返回导入数量。""" 

145 with open(filepath, "r", encoding="utf-8") as f: 

146 data = json.load(f) 

147 count = 0 

148 for name, card_data in data.items(): 

149 self.cards[name] = AgentCard.from_dict(card_data) 

150 count += 1 

151 return count 

152 

153 @classmethod 

154 def from_file(cls, filepath: str) -> "AgentCardRegistry": 

155 """从 JSON 文件创建注册中心。""" 

156 reg = cls() 

157 reg.import_from_file(filepath) 

158 return reg 

159 

160 def __len__(self) -> int: 

161 return len(self.cards) 

162 

163 def __contains__(self, name: str) -> bool: 

164 return name in self.cards 

165 

166 

167# ── AgentCardDiscovery (远程发现) ─────────────── 

168 

169class AgentCardDiscovery: 

170 """Agent Card 远程发现器。 

171 

172 通过 HTTP GET 获取远程 Agent 的 /agent-card 端点。 

173 """ 

174 

175 @staticmethod 

176 async def fetch(url: str, timeout: float = 10.0) -> Optional[AgentCard]: 

177 """从远程 URL 获取 AgentCard JSON 并解析。 

178 

179 默认期望端点返回 {"name":..., "description":..., ...} 

180 

181 Args: 

182 url: Agent Card 端点 URL（如 http://host:8000/agent-card） 

183 timeout: 请求超时（秒） 

184 

185 Returns: 

186 AgentCard 实例，失败返回 None 

187 """ 

188 try: 

189 import httpx 

190 async with httpx.AsyncClient(timeout=timeout) as client: 

191 resp = await client.get(url) 

192 resp.raise_for_status() 

193 return AgentCard.from_json(resp.text) 

194 except Exception: 

195 return None 

196 

197 @staticmethod 

198 async def fetch_all(urls: List[str], timeout: float = 10.0) -> Dict[str, Optional[AgentCard]]: 

199 """并发获取多个 Agent Card。 

200 

201 Args: 

202 urls: Agent Card 端点 URL 列表 

203 timeout: 单个请求超时（秒） 

204 

205 Returns: 

206 {url: AgentCard 或 None} 字典 

207 """ 

208 import asyncio 

209 results = await asyncio.gather( 

210 *(AgentCardDiscovery.fetch(url, timeout) for url in urls), 

211 return_exceptions=True, 

212 ) 

213 return { 

214 url: (None if isinstance(r, Exception) else r) 

215 for url, r in zip(urls, results) 

216 } 

217 

218 

219# ── 便捷函数 ─────────────────────────────────── 

220 

221def create_card( 

222 name: str, 

223 description: str, 

224 version: str = "1.0.0", 

225 url: str = "", 

226 capabilities: List[str] | None = None, 

227 skills: List[str] | None = None, 

228 **metadata, 

229) -> AgentCard: 

230 """快速创建 AgentCard 的便捷函数。""" 

231 return AgentCard( 

232 name=name, 

233 description=description, 

234 version=version, 

235 url=url, 

236 capabilities=capabilities or [], 

237 skills=skills or [], 

238 metadata=metadata, 

239 ) 

240 

241 

242def discover_local(directory: str, pattern: str = "agent-card*.json") -> List[AgentCard]: 

243 """从本地目录发现 AgentCard JSON 文件。 

244 

245 Args: 

246 directory: 扫描目录 

247 pattern: 文件名 glob pattern（仅支持简单前缀/后缀匹配） 

248 

249 Returns: 

250 发现的 AgentCard 列表 

251 """ 

252 import os 

253 import fnmatch 

254 cards: List[AgentCard] = [] 

255 try: 

256 for fname in os.listdir(directory): 

257 if fnmatch.fnmatch(fname, pattern): 

258 fpath = os.path.join(directory, fname) 

259 try: 

260 with open(fpath, "r", encoding="utf-8") as f: 

261 cards.append(AgentCard.from_json(f.read())) 

262 except Exception: 

263 continue 

264 except FileNotFoundError: 

265 pass 

266 return cards