Coverage for agentos/marketplace/manifest.py: 0%

1""" 

2AgentOS Skill Marketplace — Skill Manifest v1.0。 

3 

4兼容格式： 

5 - agentos: 原生 AgentOS Skill 格式 

6 - openclaw: OpenClaw 社区 Skill 格式（自动适配） 

7 - mcp: MCP 协议 Skill（JSON-RPC stdio/sse 代理） 

8 - generic: 通用 Python 包 Skill（无约束格式） 

9 

10参考: 

11 OpenClaw Marketplace: https://github.com/openclaw/skills 

12 MCP Specification: https://modelcontextprotocol.io 

13""" 

14 

15from __future__ import annotations 

16 

17import hashlib 

18import json 

19from dataclasses import dataclass, field 

20from enum import StrEnum 

21from pathlib import Path 

22 

23 

24class SkillFormat(StrEnum): 

25 AGENTOS = "agentos" 

26 OPENCLAW = "openclaw" 

27 MCP = "mcp" 

28 GENERIC = "generic" 

29 

30 @classmethod 

31 def detect(cls, raw: dict) -> SkillFormat: 

32 """从原始 manifest dict 自动检测格式。""" 

33 if ( 

34 raw.get("mcpServers") 

35 or raw.get("tools") 

36 and isinstance(raw.get("tools"), list) 

37 and raw["tools"] 

38 and "server" in raw["tools"][0] 

39 ): 

40 return cls.MCP 

41 if raw.get("format") == "openclaw" or raw.get("openclaw_version"): 

42 return cls.OPENCLAW 

43 if ( 

44 raw.get("format") == "agentos" 

45 or raw.get("entrypoint") 

46 or raw.get("tools") 

47 and isinstance(raw.get("tools"), list) 

48 ): 

49 return cls.AGENTOS 

50 return cls.GENERIC 

51 

52 

53@dataclass 

54class ToolDef: 

55 """Skill 暴露的工具定义。""" 

56 

57 name: str 

58 description: str = "" 

59 parameters: dict = field(default_factory=dict) 

60 returns: str = "" 

61 

62 

63@dataclass 

64class SkillManifest: 

65 """统一的 Skill 清单 — 跨格式兼容。 

66 

67 支持从 agentos / openclaw / mcp / generic 四种格式的 manifest 

68 自动解析为统一模型。安装和解依赖均基于本模型。 

69 """ 

70 

71 name: str 

72 version: str = "0.1.0" 

73 description: str = "" 

74 author: str = "unknown" 

75 license_: str = "MIT" 

76 format: SkillFormat = SkillFormat.GENERIC 

77 

78 # AgentOS 原生字段 

79 entrypoint: str = "" # "module:func" 格式的入口点 

80 tools: list[ToolDef] = field(default_factory=list) 

81 dependencies: list[str] = field(default_factory=list) 

82 

83 # MCP 兼容字段 

84 mcp_command: str = "" # MCP server 启动命令，如 "npx -y @anthropic/mcp-server" 

85 mcp_args: list[str] = field(default_factory=list) 

86 mcp_env: dict = field(default_factory=dict) 

87 mcp_type: str = "stdio" # stdio | sse 

88 

89 # OpenClaw 兼容字段 

90 openclaw_version: str = "" 

91 

92 # 通用字段 

93 tags: list[str] = field(default_factory=list) 

94 homepage: str = "" 

95 repository: str = "" 

96 icon: str = "" 

97 min_agentos_version: str = "1.7.0" 

98 

99 # 元数据 

100 install_path: str = "" 

101 source: str = "" # pypi | github | local | url 

102 manifest_hash: str = "" 

103 

104 @classmethod 

105 def from_dict(cls, raw: dict, source: str = "", install_path: str = "") -> SkillManifest: 

106 """从原始 dict 自动检测格式并解析。""" 

107 fmt = SkillFormat.detect(raw) 

108 m = cls(name="", description="") 

109 

110 m.name = raw.get("name", "") 

111 m.version = str(raw.get("version", "0.1.0")) 

112 m.description = raw.get("description", "") 

113 m.author = raw.get("author", raw.get("maintainer", "unknown")) 

114 m.license_ = raw.get("license", raw.get("license_", "MIT")) 

115 m.format = fmt 

116 m.source = source 

117 m.install_path = install_path 

118 m.tags = raw.get("tags", raw.get("keywords", [])) 

119 m.homepage = raw.get("homepage", raw.get("url", "")) 

120 m.repository = raw.get("repository", raw.get("repo", "")) 

121 m.icon = raw.get("icon", "") 

122 m.min_agentos_version = raw.get("min_agentos_version", raw.get("requires_agentos", "1.7.0")) 

123 

124 if fmt == SkillFormat.AGENTOS: 

125 m.entrypoint = raw.get("entrypoint", "") 

126 m.dependencies = raw.get("dependencies", raw.get("requires", [])) 

127 tools_raw = raw.get("tools", []) 

128 for t in tools_raw: 

129 m.tools.append( 

130 ToolDef( 

131 name=t.get("name", ""), 

132 description=t.get("description", ""), 

133 parameters=t.get("parameters", {}), 

134 returns=t.get("returns", ""), 

135 ) 

136 ) 

137 

138 elif fmt == SkillFormat.OPENCLAW: 

139 # OpenClaw 格式：skill.yaml → agentos 适配 

140 m.entrypoint = raw.get("entrypoint", raw.get("main", "")) 

141 m.dependencies = raw.get("dependencies", raw.get("pip", [])) 

142 m.openclaw_version = raw.get("openclaw_version", raw.get("format_version", "")) 

143 tools_raw = raw.get("tools", raw.get("functions", [])) 

144 for t in tools_raw: 

145 m.tools.append( 

146 ToolDef( 

147 name=t.get("name", ""), 

148 description=t.get("description", ""), 

149 parameters=t.get("parameters", t.get("input_schema", {})), 

150 ) 

151 ) 

152 

153 elif fmt == SkillFormat.MCP: 

154 # MCP 格式：mcpServers.{name} → agentos 适配 

155 servers = raw.get("mcpServers", {}) 

156 if servers: 

157 first = list(servers.values())[0] if servers else {} 

158 m.mcp_command = first.get("command", "") 

159 m.mcp_args = first.get("args", []) 

160 m.mcp_env = first.get("env", {}) 

161 m.mcp_type = first.get("type", "stdio") 

162 if not m.name and "server_name" in raw: 

163 m.name = raw["server_name"] 

164 if not m.description: 

165 m.description = f"MCP Server: {m.mcp_command} {' '.join(m.mcp_args)}" 

166 tools_raw = raw.get("tools", []) 

167 for t in tools_raw: 

168 m.tools.append( 

169 ToolDef( 

170 name=t.get("name", ""), 

171 description=t.get("description", ""), 

172 parameters=t.get("inputSchema", {}), 

173 ) 

174 ) 

175 

176 elif fmt == SkillFormat.GENERIC: 

177 m.entrypoint = raw.get("entrypoint", raw.get("main", "")) 

178 m.dependencies = raw.get( 

179 "dependencies", raw.get("requires", raw.get("install_requires", [])) 

180 ) 

181 m.description = raw.get("description", raw.get("summary", "")) 

182 

183 # 计算 manifest 哈希 

184 m.manifest_hash = m._compute_hash() 

185 return m 

186 

187 def to_dict(self) -> dict: 

188 """导出为标准 agentos manifest dict。""" 

189 return { 

190 "name": self.name, 

191 "version": self.version, 

192 "description": self.description, 

193 "author": self.author, 

194 "license": self.license_, 

195 "format": self.format.value, 

196 "entrypoint": self.entrypoint, 

197 "tools": [ 

198 { 

199 "name": t.name, 

200 "description": t.description, 

201 "parameters": t.parameters, 

202 "returns": t.returns, 

203 } 

204 for t in self.tools 

205 ], 

206 "dependencies": self.dependencies, 

207 "tags": self.tags, 

208 "homepage": self.homepage, 

209 "repository": self.repository, 

210 "icon": self.icon, 

211 "min_agentos_version": self.min_agentos_version, 

212 "mcp": ( 

213 { 

214 "command": self.mcp_command, 

215 "args": self.mcp_args, 

216 "env": self.mcp_env, 

217 "type": self.mcp_type, 

218 } 

219 if self.mcp_command 

220 else None 

221 ), 

222 "manifest_hash": self.manifest_hash, 

223 "source": self.source, 

224 } 

225 

226 def _compute_hash(self) -> str: 

227 raw = json.dumps(self.to_dict(), sort_keys=True, default=str) 

228 return hashlib.sha256(raw.encode()).hexdigest()[:16] 

229 

230 @staticmethod 

231 def load_from_path(manifest_path: str | Path, source: str = "local") -> SkillManifest | None: 

232 """从本地 manifest 文件加载。支持 yaml/json。""" 

233 p = Path(manifest_path) 

234 if not p.exists(): 

235 return None 

236 text = p.read_text(encoding="utf-8") 

237 if p.suffix in (".yaml", ".yml"): 

238 import yaml 

239 

240 raw = yaml.safe_load(text) 

241 else: 

242 raw = json.loads(text) 

243 return SkillManifest.from_dict(raw, source=source, install_path=str(p.parent))