Coverage for integrations / openclaw / clawhub_adapter.py: 73.9%
226 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2ClawHub Adapter — Parse, install, and execute OpenClaw skills in HART OS.
4ClawHub skills are simple: a SKILL.md with YAML frontmatter + instructions.
5We parse them into HART-compatible actions that agents can invoke.
7Flow:
8 1. `clawhub install <slug>` downloads skill to ~/.hevolve/openclaw/skills/
9 2. SkillParser reads SKILL.md frontmatter (name, description, metadata.openclaw)
10 3. Skill requirements are checked (bins, env vars, config)
11 4. Instructions body is loaded as agent context (like a mini recipe)
12 5. HART agents can invoke the skill as a tool via ClawHubToolProvider
13"""
15import json
16import logging
17import os
18import re
19import shutil
20import subprocess
21from dataclasses import dataclass, field
22from pathlib import Path
23from typing import Any, Dict, List, Optional
25logger = logging.getLogger(__name__)
27# ── Paths ──────────────────────────────────────────────────────────
29OPENCLAW_HOME = Path(os.environ.get(
30 'OPENCLAW_SKILLS_DIR',
31 os.path.expanduser('~/.hevolve/openclaw/skills')
32))
34CLAWHUB_REGISTRY = 'https://registry.clawhub.ai'
37# ── Skill Schema ───────────────────────────────────────────────────
39@dataclass
40class OpenClawRequirements:
41 """Requirements declared in metadata.openclaw.requires."""
42 bins: List[str] = field(default_factory=list)
43 any_bins: List[str] = field(default_factory=list)
44 env: List[str] = field(default_factory=list)
45 config: List[str] = field(default_factory=list)
48@dataclass
49class OpenClawInstallSpec:
50 """An install directive (brew, node, go, uv, nix)."""
51 id: str = ''
52 kind: str = '' # brew, node, go, uv, nix, pip
53 formula: str = '' # package name
54 bins: List[str] = field(default_factory=list)
55 label: str = ''
56 os_filter: List[str] = field(default_factory=lambda: ['all'])
59@dataclass
60class OpenClawSkill:
61 """Parsed representation of a SKILL.md file."""
62 name: str = ''
63 description: str = ''
64 version: str = '0.0.0'
65 homepage: str = ''
66 emoji: str = ''
67 user_invocable: bool = True
68 disable_model_invocation: bool = False
69 command_dispatch: Optional[str] = None # 'tool' or None
70 command_tool: Optional[str] = None
71 command_arg_mode: str = 'raw'
72 primary_env: str = ''
73 os_filter: List[str] = field(default_factory=lambda: ['all'])
74 requirements: OpenClawRequirements = field(default_factory=OpenClawRequirements)
75 install_specs: List[OpenClawInstallSpec] = field(default_factory=list)
76 instructions: str = '' # The body of SKILL.md
77 skill_dir: str = '' # Local path
78 source: str = 'clawhub' # 'clawhub', 'local', 'workspace'
81# ── Parser ─────────────────────────────────────────────────────────
83_FRONTMATTER_RE = re.compile(r'^---\s*\n(.*?)\n---\s*\n', re.DOTALL)
86def _parse_yaml_simple(text: str) -> Dict[str, Any]:
87 """Minimal YAML-like parser for SKILL.md frontmatter.
89 ClawHub frontmatter is constrained: single-line keys, JSON metadata.
90 We avoid a PyYAML dependency by parsing the subset we need.
91 """
92 result = {}
93 for line in text.split('\n'):
94 line = line.strip()
95 if not line or line.startswith('#'):
96 continue
97 colon_idx = line.find(':')
98 if colon_idx < 0:
99 continue
100 key = line[:colon_idx].strip()
101 value = line[colon_idx + 1:].strip()
102 # Try JSON parse for metadata objects
103 if value.startswith('{') or value.startswith('['):
104 try:
105 value = json.loads(value)
106 except json.JSONDecodeError:
107 pass
108 elif value.lower() in ('true', 'yes'):
109 value = True
110 elif value.lower() in ('false', 'no'):
111 value = False
112 elif value.isdigit():
113 value = int(value)
114 # Strip quotes
115 elif isinstance(value, str) and len(value) >= 2:
116 if (value[0] == '"' and value[-1] == '"') or \
117 (value[0] == "'" and value[-1] == "'"):
118 value = value[1:-1]
119 result[key] = value
120 return result
123def parse_skill_md(skill_path: str) -> OpenClawSkill:
124 """Parse a SKILL.md file into an OpenClawSkill object."""
125 path = Path(skill_path)
126 if path.is_dir():
127 path = path / 'SKILL.md'
129 content = path.read_text(encoding='utf-8')
130 skill = OpenClawSkill(skill_dir=str(path.parent))
132 # Parse frontmatter
133 fm_match = _FRONTMATTER_RE.match(content)
134 if fm_match:
135 fm = _parse_yaml_simple(fm_match.group(1))
136 skill.name = fm.get('name', '')
137 skill.description = fm.get('description', '')
138 skill.version = str(fm.get('version', '0.0.0'))
139 skill.homepage = fm.get('homepage', '')
140 skill.user_invocable = fm.get('user-invocable', True)
141 skill.disable_model_invocation = fm.get('disable-model-invocation', False)
142 skill.command_dispatch = fm.get('command-dispatch')
143 skill.command_tool = fm.get('command-tool')
144 skill.command_arg_mode = fm.get('command-arg-mode', 'raw')
146 # Parse metadata.openclaw
147 meta = fm.get('metadata', {})
148 if isinstance(meta, str):
149 try:
150 meta = json.loads(meta)
151 except json.JSONDecodeError:
152 meta = {}
153 oc = meta.get('openclaw', {}) if isinstance(meta, dict) else {}
155 skill.emoji = oc.get('emoji', '')
156 skill.primary_env = oc.get('primaryEnv', '')
157 skill.os_filter = oc.get('os', ['all'])
159 # Requirements
160 req = oc.get('requires', {})
161 if isinstance(req, dict):
162 skill.requirements = OpenClawRequirements(
163 bins=req.get('bins', []),
164 any_bins=req.get('anyBins', []),
165 env=req.get('env', []),
166 config=req.get('config', []),
167 )
169 # Install specs
170 for spec_data in oc.get('install', []):
171 if isinstance(spec_data, dict):
172 skill.install_specs.append(OpenClawInstallSpec(
173 id=spec_data.get('id', ''),
174 kind=spec_data.get('kind', ''),
175 formula=spec_data.get('formula', ''),
176 bins=spec_data.get('bins', []),
177 label=spec_data.get('label', ''),
178 os_filter=spec_data.get('os', ['all']),
179 ))
181 # Instructions = everything after frontmatter
182 skill.instructions = content[fm_match.end():].strip()
183 else:
184 # No frontmatter — entire file is instructions
185 skill.instructions = content.strip()
187 return skill
190# ── Requirements Check ─────────────────────────────────────────────
192def check_requirements(skill: OpenClawSkill) -> Dict[str, Any]:
193 """Check if skill requirements are satisfied on this system.
195 Returns dict with 'satisfied' bool and 'missing' details.
196 """
197 missing_bins = []
198 missing_env = []
199 req = skill.requirements
201 for b in req.bins:
202 if not shutil.which(b):
203 missing_bins.append(b)
205 if req.any_bins:
206 if not any(shutil.which(b) for b in req.any_bins):
207 missing_bins.append(f"any of: {', '.join(req.any_bins)}")
209 for e in req.env:
210 if not os.environ.get(e):
211 missing_env.append(e)
213 satisfied = len(missing_bins) == 0 and len(missing_env) == 0
214 return {
215 'satisfied': satisfied,
216 'missing_bins': missing_bins,
217 'missing_env': missing_env,
218 }
221# ── Install / Uninstall ───────────────────────────────────────────
223def install_skill(slug: str, version: Optional[str] = None,
224 force: bool = False) -> Optional[OpenClawSkill]:
225 """Install a ClawHub skill by slug.
227 Downloads from registry or uses clawhub CLI if available.
228 Skills are stored in ~/.hevolve/openclaw/skills/<slug>/
229 """
230 dest = OPENCLAW_HOME / slug
231 if dest.exists() and not force:
232 logger.info("Skill %s already installed at %s", slug, dest)
233 return parse_skill_md(str(dest))
235 dest.mkdir(parents=True, exist_ok=True)
237 # Strategy 1: Use clawhub CLI if available
238 clawhub_bin = shutil.which('clawhub')
239 if clawhub_bin:
240 cmd = [clawhub_bin, 'install', slug]
241 if version:
242 cmd.extend(['--version', version])
243 if force:
244 cmd.append('--force')
245 try:
246 result = subprocess.run(
247 cmd, capture_output=True, text=True, timeout=60,
248 env={**os.environ, 'CLAWHUB_SKILLS_DIR': str(OPENCLAW_HOME)}
249 )
250 if result.returncode == 0:
251 logger.info("Installed skill %s via clawhub CLI", slug)
252 return parse_skill_md(str(dest))
253 logger.warning("clawhub install failed: %s", result.stderr)
254 except (subprocess.TimeoutExpired, FileNotFoundError) as e:
255 logger.warning("clawhub CLI error: %s", e)
257 # Strategy 2: Direct HTTP download from registry
258 try:
259 from core.http_pool import pooled_get
260 except ImportError:
261 import requests
262 pooled_get = requests.get
264 url = f"{CLAWHUB_REGISTRY}/api/skills/{slug}"
265 if version:
266 url += f"/versions/{version}"
267 url += "/download"
269 try:
270 resp = pooled_get(url, timeout=30)
271 if hasattr(resp, 'status_code'):
272 status = resp.status_code
273 else:
274 status = getattr(resp, 'status', 200)
276 if status == 200:
277 # Registry returns a tarball or SKILL.md content
278 content_type = ''
279 if hasattr(resp, 'headers'):
280 content_type = resp.headers.get('content-type', '')
282 if 'application/json' in content_type:
283 data = resp.json() if hasattr(resp, 'json') else json.loads(resp.text)
284 skill_md = data.get('skill_md', data.get('content', ''))
285 (dest / 'SKILL.md').write_text(skill_md, encoding='utf-8')
286 else:
287 text = resp.text if hasattr(resp, 'text') else str(resp)
288 (dest / 'SKILL.md').write_text(text, encoding='utf-8')
290 logger.info("Installed skill %s from registry", slug)
291 return parse_skill_md(str(dest))
292 else:
293 logger.error("Registry returned %d for skill %s", status, slug)
294 except Exception as e:
295 logger.error("Failed to download skill %s: %s", slug, e)
297 # Clean up empty dir on failure
298 if dest.exists() and not any(dest.iterdir()):
299 dest.rmdir()
300 return None
303def uninstall_skill(slug: str) -> bool:
304 """Remove an installed skill."""
305 dest = OPENCLAW_HOME / slug
306 if dest.exists():
307 shutil.rmtree(dest)
308 logger.info("Uninstalled skill %s", slug)
309 return True
310 return False
313def list_installed_skills() -> List[OpenClawSkill]:
314 """List all installed OpenClaw skills."""
315 skills = []
316 if not OPENCLAW_HOME.exists():
317 return skills
318 for d in sorted(OPENCLAW_HOME.iterdir()):
319 skill_md = d / 'SKILL.md'
320 if d.is_dir() and skill_md.exists():
321 try:
322 skills.append(parse_skill_md(str(d)))
323 except Exception as e:
324 logger.warning("Failed to parse skill at %s: %s", d, e)
325 return skills
328# ── Skill → HART Tool Conversion ──────────────────────────────────
330def skill_to_autogen_tool(skill: OpenClawSkill) -> Dict[str, Any]:
331 """Convert an OpenClaw skill to an AutoGen-compatible tool definition.
333 The tool wraps the skill's instructions as a prompt-based action
334 that HART agents can invoke like any other tool.
335 """
336 tool_name = f"openclaw_{skill.name.replace('-', '_')}"
338 def tool_fn(command: str = '', **kwargs) -> str:
339 """Execute an OpenClaw skill with the given command/args."""
340 # Replace {baseDir} placeholder with actual skill dir
341 instructions = skill.instructions.replace('{baseDir}', skill.skill_dir)
343 # If command-dispatch: tool, forward directly
344 if skill.command_dispatch == 'tool' and skill.command_tool:
345 return json.dumps({
346 'tool': skill.command_tool,
347 'command': command,
348 'skill': skill.name,
349 'instructions': instructions,
350 })
352 # Otherwise, return instructions as context for the agent
353 return json.dumps({
354 'skill': skill.name,
355 'description': skill.description,
356 'instructions': instructions,
357 'command': command,
358 })
360 return {
361 'name': tool_name,
362 'description': skill.description or f"OpenClaw skill: {skill.name}",
363 'function': tool_fn,
364 'parameters': {
365 'command': {
366 'type': 'string',
367 'description': 'Command or input to pass to the skill',
368 }
369 },
370 'source': 'openclaw_clawhub',
371 }
374class ClawHubToolProvider:
375 """Provides all installed ClawHub skills as HART agent tools.
377 Usage:
378 provider = ClawHubToolProvider()
379 tools = provider.get_tools()
380 # Register with AutoGen agent
381 for tool in tools:
382 agent.register_function(tool['function'], tool['name'], tool['description'])
383 """
385 def __init__(self, skills_dir: Optional[str] = None):
386 self._skills_dir = Path(skills_dir) if skills_dir else OPENCLAW_HOME
387 self._cache: Optional[List[Dict[str, Any]]] = None
389 def get_tools(self, refresh: bool = False) -> List[Dict[str, Any]]:
390 """Get all installed skills as AutoGen tool definitions."""
391 if self._cache is not None and not refresh:
392 return self._cache
394 tools = []
395 if self._skills_dir.exists():
396 for d in sorted(self._skills_dir.iterdir()):
397 skill_md = d / 'SKILL.md'
398 if d.is_dir() and skill_md.exists():
399 try:
400 skill = parse_skill_md(str(d))
401 if skill.disable_model_invocation:
402 continue
403 req_check = check_requirements(skill)
404 if not req_check['satisfied']:
405 logger.debug("Skipping skill %s: missing %s",
406 skill.name, req_check)
407 continue
408 tools.append(skill_to_autogen_tool(skill))
409 except Exception as e:
410 logger.warning("Failed to load skill %s: %s", d.name, e)
412 self._cache = tools
413 return tools
415 def invalidate(self):
416 """Clear cached tools (call after install/uninstall)."""
417 self._cache = None
420# ── Singleton ──────────────────────────────────────────────────────
422_provider: Optional[ClawHubToolProvider] = None
425def get_clawhub_provider() -> ClawHubToolProvider:
426 """Get the singleton ClawHub tool provider."""
427 global _provider
428 if _provider is None:
429 _provider = ClawHubToolProvider()
430 return _provider