Coverage for integrations / openclaw / hart_skill_server.py: 53.7%

1""" 

2HART Skill Server — Expose HART OS agents/recipes as OpenClaw-compatible tools. 

3 

4When OpenClaw is running alongside HART OS, this server makes HART capabilities 

5available as OpenClaw tools. OpenClaw agents can then: 

6 - Execute HART recipes (CREATE/REUSE) 

7 - Query HART agents 

8 - Use Model Bus for inference 

9 - Access compute mesh 

10 - Use vision/TTS services 

11 

12The server registers as an OpenClaw tool provider via the gateway. 

13""" 

14 

15import json 

16import logging 

17import os 

18from typing import Any, Dict, List, Optional 

19 

20logger = logging.getLogger(__name__) 

21 

22# HART capabilities exposed to OpenClaw 

23HART_TOOLS = { 

24 'hart_chat': { 

25 'name': 'hart_chat', 

26 'description': 'Send a task to HART OS agent for execution', 

27 'parameters': { 

28 'prompt': {'type': 'string', 'description': 'The task to execute'}, 

29 'user_id': {'type': 'string', 'description': 'User ID', 'default': '1'}, 

30 'create_agent': {'type': 'boolean', 'description': 'Create new agent', 'default': False}, 

31 }, 

32 }, 

33 'hart_recipe_run': { 

34 'name': 'hart_recipe_run', 

35 'description': 'Execute a trained HART recipe (fast, no LLM calls)', 

36 'parameters': { 

37 'prompt_id': {'type': 'string', 'description': 'Recipe prompt ID'}, 

38 'flow_id': {'type': 'string', 'description': 'Flow ID', 'default': '0'}, 

39 }, 

40 }, 

41 'hart_model_bus': { 

42 'name': 'hart_model_bus', 

43 'description': 'Run inference via HART Model Bus (local or cloud models)', 

44 'parameters': { 

45 'prompt': {'type': 'string', 'description': 'Inference prompt'}, 

46 'model': {'type': 'string', 'description': 'Model name', 'default': 'auto'}, 

47 }, 

48 }, 

49 'hart_tts': { 

50 'name': 'hart_tts', 

51 'description': 'Generate speech using HART TTS (LuxTTS/Pocket TTS, offline)', 

52 'parameters': { 

53 'text': {'type': 'string', 'description': 'Text to speak'}, 

54 'voice': {'type': 'string', 'description': 'Voice name', 'default': 'alba'}, 

55 }, 

56 }, 

57 'hart_vision': { 

58 'name': 'hart_vision', 

59 'description': 'Analyze image using HART vision (MiniCPM)', 

60 'parameters': { 

61 'image_path': {'type': 'string', 'description': 'Path to image'}, 

62 'question': {'type': 'string', 'description': 'Question about the image'}, 

63 }, 

64 }, 

65 'hart_expert': { 

66 'name': 'hart_expert', 

67 'description': 'Query HART expert agents network (96 specialists)', 

68 'parameters': { 

69 'domain': {'type': 'string', 'description': 'Expert domain'}, 

70 'query': {'type': 'string', 'description': 'Query for the expert'}, 

71 }, 

72 }, 

73} 

74 

75 

76class HARTToolHandler: 

77 """Handles tool invocations from OpenClaw by dispatching to HART services.""" 

78 

79 def __init__(self, backend_url: str = 'http://localhost:6777'): 

80 self._backend_url = backend_url 

81 

82 def handle(self, tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]: 

83 """Dispatch a tool call to the appropriate HART service.""" 

84 handler = getattr(self, f'_handle_{tool_name}', None) 

85 if handler: 

86 return handler(args) 

87 return {'error': f'Unknown tool: {tool_name}'} 

88 

89 def _handle_hart_chat(self, args: Dict[str, Any]) -> Dict[str, Any]: 

90 try: 

91 from core.http_pool import pooled_post 

92 except ImportError: 

93 import requests 

94 pooled_post = requests.post 

95 

96 resp = pooled_post( 

97 f"{self._backend_url}/chat", 

98 json={ 

99 'user_id': args.get('user_id', '1'), 

100 'prompt_id': args.get('prompt_id', '99999'), 

101 'prompt': args['prompt'], 

102 'create_agent': args.get('create_agent', False), 

103 }, 

104 timeout=120, 

105 ) 

106 if hasattr(resp, 'json'): 

107 return resp.json() 

108 return {'result': str(resp)} 

109 

110 def _handle_hart_recipe_run(self, args: Dict[str, Any]) -> Dict[str, Any]: 

111 try: 

112 from reuse_recipe import reuse_recipe 

113 result = reuse_recipe( 

114 prompt_id=args['prompt_id'], 

115 flow_id=args.get('flow_id', '0'), 

116 ) 

117 return {'result': result} 

118 except Exception as e: 

119 return {'error': str(e)} 

120 

121 def _handle_hart_model_bus(self, args: Dict[str, Any]) -> Dict[str, Any]: 

122 try: 

123 from integrations.agent_engine.model_bus_service import get_model_bus 

124 bus = get_model_bus() 

125 result = bus.infer(args['prompt'], model=args.get('model', 'auto')) 

126 return {'result': result} 

127 except Exception as e: 

128 return {'error': str(e)} 

129 

130 def _handle_hart_tts(self, args: Dict[str, Any]) -> Dict[str, Any]: 

131 try: 

132 from integrations.audio.tts import get_tts_engine 

133 engine = get_tts_engine() 

134 path = engine.synthesize(args['text'], voice=args.get('voice', 'alba')) 

135 return {'audio_path': path} 

136 except Exception as e: 

137 return {'error': str(e)} 

138 

139 def _handle_hart_vision(self, args: Dict[str, Any]) -> Dict[str, Any]: 

140 try: 

141 from integrations.vision.vision_service import get_vision_service 

142 svc = get_vision_service() 

143 result = svc.analyze(args['image_path'], args.get('question', 'Describe this image')) 

144 return {'result': result} 

145 except Exception as e: 

146 return {'error': str(e)} 

147 

148 def _handle_hart_expert(self, args: Dict[str, Any]) -> Dict[str, Any]: 

149 try: 

150 from integrations.expert_agents.agent_network import get_expert_network 

151 network = get_expert_network() 

152 result = network.query(args['domain'], args['query']) 

153 return {'result': result} 

154 except Exception as e: 

155 return {'error': str(e)} 

156 

157 def list_tools(self) -> List[Dict[str, Any]]: 

158 """List all HART tools available to OpenClaw.""" 

159 return list(HART_TOOLS.values()) 

160 

161 

162def generate_hart_skills() -> List[str]: 

163 """Generate SKILL.md files for all HART tools. 

164 

165 These can be installed in OpenClaw to give it native HART access. 

166 Returns list of generated SKILL.md content strings. 

167 """ 

168 skills = [] 

169 for tool_name, tool_def in HART_TOOLS.items(): 

170 params_doc = [] 

171 for pname, pinfo in tool_def.get('parameters', {}).items(): 

172 default = pinfo.get('default', '') 

173 params_doc.append( 

174 f"- `{pname}` ({pinfo['type']}): {pinfo['description']}" 

175 + (f" (default: {default})" if default != '' else '') 

176 ) 

177 

178 skill_md = f"""--- 

179name: {tool_name} 

180description: {tool_def['description']} 

181version: 1.0.0 

182homepage: https://github.com/hertz-ai/HARTOS 

183metadata: {{"openclaw": {{"emoji": "\\U0001f916", "requires": {{"env": ["HART_BACKEND_URL"]}}}}}} 

184--- 

185 

186# {tool_def['name']} 

187 

188{tool_def['description']} 

189 

190## Usage 

191 

192This tool connects to a running HART OS instance. Set `HART_BACKEND_URL` 

193to the HART backend address (default: http://localhost:6777). 

194 

195## Parameters 

196 

197{chr(10).join(params_doc)} 

198 

199## How it works 

200 

2011. Receives the request from OpenClaw 

2022. Forwards to HART OS backend via HTTP 

2033. Returns the result to the OpenClaw agent 

204 

205This is a bridge skill — the actual work is done by HART OS services. 

206""" 

207 skills.append(skill_md) 

208 return skills 

209 

210 

211# ── Singleton ────────────────────────────────────────────────────── 

212 

213_handler: Optional[HARTToolHandler] = None 

214 

215 

216def get_hart_tool_handler() -> HARTToolHandler: 

217 """Get the singleton HART tool handler.""" 

218 global _handler 

219 if _handler is None: 

220 _handler = HARTToolHandler() 

221 return _handler