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 asdict, dataclass, field 

17from typing import Any 

18 

19# ── AgentCard ─────────────────────────────────── 

20 

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 

97@dataclass 

98class AgentCardRegistry: 

99 """Agent Card 注册中心。 

100 

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

102 """ 

103 

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

105 

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

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

108 self.cards[card.name] = card 

109 

110 def unregister(self, name: str) -> AgentCard | None: 

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

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

113 

114 def get(self, name: str) -> AgentCard | None: 

115 """按名称查找。""" 

116 return self.cards.get(name) 

117 

118 def list_all(self) -> list[AgentCard]: 

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

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

121 

122 def find_by_query(self, query: str) -> list[AgentCard]: 

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

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

125 

126 def find_by_capability(self, capability: str) -> list[AgentCard]: 

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

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

129 

130 def find_by_skill(self, skill: str) -> list[AgentCard]: 

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

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

133 

134 def find_by_tag(self, tag: str) -> list[AgentCard]: 

135 """按标签查找。""" 

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

137 

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

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

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

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

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

143 

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

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

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

147 data = json.load(f) 

148 count = 0 

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

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

151 count += 1 

152 return count 

153 

154 @classmethod 

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

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

157 reg = cls() 

158 reg.import_from_file(filepath) 

159 return reg 

160 

161 def __len__(self) -> int: 

162 return len(self.cards) 

163 

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

165 return name in self.cards 

166 

167 

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

169 

170 

171class AgentCardDiscovery: 

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

173 

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

175 """ 

176 

177 @staticmethod 

178 async def fetch(url: str, timeout: float = 10.0) -> AgentCard | None: 

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

180 

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

182 

183 Args: 

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

185 timeout: 请求超时（秒） 

186 

187 Returns: 

188 AgentCard 实例，失败返回 None 

189 """ 

190 try: 

191 import httpx 

192 

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

194 resp = await client.get(url) 

195 resp.raise_for_status() 

196 return AgentCard.from_json(resp.text) 

197 except Exception: 

198 return None 

199 

200 @staticmethod 

201 async def fetch_all(urls: list[str], timeout: float = 10.0) -> dict[str, AgentCard | None]: 

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

203 

204 Args: 

205 urls: Agent Card 端点 URL 列表 

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

207 

208 Returns: 

209 {url: AgentCard 或 None} 字典 

210 """ 

211 import asyncio 

212 

213 results = await asyncio.gather( 

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

215 return_exceptions=True, 

216 ) 

217 return {url: (None if isinstance(r, Exception) else r) for url, r in zip(urls, results)} 

218 

219 

220# ── 便捷函数 ─────────────────────────────────── 

221 

222 

223def create_card( 

224 name: str, 

225 description: str, 

226 version: str = "1.0.0", 

227 url: str = "", 

228 capabilities: list[str] | None = None, 

229 skills: list[str] | None = None, 

230 **metadata, 

231) -> AgentCard: 

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

233 return AgentCard( 

234 name=name, 

235 description=description, 

236 version=version, 

237 url=url, 

238 capabilities=capabilities or [], 

239 skills=skills or [], 

240 metadata=metadata, 

241 ) 

242 

243 

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

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

246 

247 Args: 

248 directory: 扫描目录 

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

250 

251 Returns: 

252 发现的 AgentCard 列表 

253 """ 

254 import fnmatch 

255 import os 

256 

257 cards: list[AgentCard] = [] 

258 try: 

259 for fname in os.listdir(directory): 

260 if fnmatch.fnmatch(fname, pattern): 

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

262 try: 

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

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

265 except Exception: 

266 continue 

267 except FileNotFoundError: 

268 pass 

269 return cards