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 Enum 

21from pathlib import Path 

22from typing import Optional 

23 

24 

25class SkillFormat(str, Enum): 

26 AGENTOS = "agentos" 

27 OPENCLAW = "openclaw" 

28 MCP = "mcp" 

29 GENERIC = "generic" 

30 

31 @classmethod 

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

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

34 if raw.get("mcpServers") or raw.get("tools") and isinstance(raw.get("tools"), list) and raw["tools"] and "server" in raw["tools"][0]: 

35 return cls.MCP 

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

37 return cls.OPENCLAW 

38 if raw.get("format") == "agentos" or raw.get("entrypoint") or raw.get("tools") and isinstance(raw.get("tools"), list): 

39 return cls.AGENTOS 

40 return cls.GENERIC 

41 

42 

43@dataclass 

44class ToolDef: 

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

46 name: str 

47 description: str = "" 

48 parameters: dict = field(default_factory=dict) 

49 returns: str = "" 

50 

51 

52@dataclass 

53class SkillManifest: 

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

55 

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

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

58 """ 

59 

60 name: str 

61 version: str = "0.1.0" 

62 description: str = "" 

63 author: str = "unknown" 

64 license_: str = "MIT" 

65 format: SkillFormat = SkillFormat.GENERIC 

66 

67 # AgentOS 原生字段 

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

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

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

71 

72 # MCP 兼容字段 

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

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

75 mcp_env: dict = field(default_factory=dict) 

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

77 

78 # OpenClaw 兼容字段 

79 openclaw_version: str = "" 

80 

81 # 通用字段 

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

83 homepage: str = "" 

84 repository: str = "" 

85 icon: str = "" 

86 min_agentos_version: str = "1.7.0" 

87 

88 # 元数据 

89 install_path: str = "" 

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

91 manifest_hash: str = "" 

92 

93 @classmethod 

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

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

96 fmt = SkillFormat.detect(raw) 

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

98 

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

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

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

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

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

104 m.format = fmt 

105 m.source = source 

106 m.install_path = install_path 

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

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

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

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

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

112 

113 if fmt == SkillFormat.AGENTOS: 

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

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

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

117 for t in tools_raw: 

118 m.tools.append(ToolDef( 

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

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

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

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

123 )) 

124 

125 elif fmt == SkillFormat.OPENCLAW: 

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

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

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

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

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

131 for t in tools_raw: 

132 m.tools.append(ToolDef( 

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

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

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

136 )) 

137 

138 elif fmt == SkillFormat.MCP: 

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

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

141 if servers: 

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

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

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

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

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

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

148 m.name = raw["server_name"] 

149 if not m.description: 

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

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

152 for t in tools_raw: 

153 m.tools.append(ToolDef( 

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

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

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

157 )) 

158 

159 elif fmt == SkillFormat.GENERIC: 

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

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

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

163 

164 # 计算 manifest 哈希 

165 m.manifest_hash = m._compute_hash() 

166 return m 

167 

168 def to_dict(self) -> dict: 

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

170 return { 

171 "name": self.name, 

172 "version": self.version, 

173 "description": self.description, 

174 "author": self.author, 

175 "license": self.license_, 

176 "format": self.format.value, 

177 "entrypoint": self.entrypoint, 

178 "tools": [ 

179 { 

180 "name": t.name, 

181 "description": t.description, 

182 "parameters": t.parameters, 

183 "returns": t.returns, 

184 } 

185 for t in self.tools 

186 ], 

187 "dependencies": self.dependencies, 

188 "tags": self.tags, 

189 "homepage": self.homepage, 

190 "repository": self.repository, 

191 "icon": self.icon, 

192 "min_agentos_version": self.min_agentos_version, 

193 "mcp": { 

194 "command": self.mcp_command, 

195 "args": self.mcp_args, 

196 "env": self.mcp_env, 

197 "type": self.mcp_type, 

198 } if self.mcp_command else None, 

199 "manifest_hash": self.manifest_hash, 

200 "source": self.source, 

201 } 

202 

203 def _compute_hash(self) -> str: 

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

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

206 

207 @staticmethod 

208 def load_from_path(manifest_path: str | Path, source: str = "local") -> Optional["SkillManifest"]: 

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

210 p = Path(manifest_path) 

211 if not p.exists(): 

212 return None 

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

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

215 import yaml 

216 raw = yaml.safe_load(text) 

217 else: 

218 raw = json.loads(text) 

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