Coverage for integrations / openclaw / shell_openclaw_apis.py: 61.0%
100 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Shell OpenClaw APIs — REST endpoints for OpenClaw integration in Nunba.
4Provides:
5 - /api/openclaw/skills — List installed ClawHub skills
6 - /api/openclaw/skills/install — Install a skill by slug
7 - /api/openclaw/skills/uninstall — Remove a skill
8 - /api/openclaw/skills/search — Search ClawHub registry
9 - /api/openclaw/status — OpenClaw gateway status
10 - /api/openclaw/channels — Available messaging channels
11 - /api/assistant/chat — Floating assistant chat endpoint
12 - /api/assistant/capabilities — What the assistant can do
13"""
15import json
16import logging
17from typing import Any, Dict
19logger = logging.getLogger(__name__)
22def register_openclaw_routes(app):
23 """Register OpenClaw and floating assistant routes on a Flask app."""
25 # ── OpenClaw Skill Management ──────────────────────────────
27 @app.route('/api/openclaw/skills', methods=['GET'])
28 def _openclaw_list_skills():
29 from flask import jsonify
30 try:
31 from integrations.openclaw.clawhub_adapter import list_installed_skills
32 skills = list_installed_skills()
33 return jsonify({
34 'success': True,
35 'skills': [
36 {
37 'name': s.name,
38 'description': s.description,
39 'version': s.version,
40 'emoji': s.emoji,
41 'homepage': s.homepage,
42 'user_invocable': s.user_invocable,
43 'requirements_met': True, # Already filtered
44 }
45 for s in skills
46 ],
47 'count': len(skills),
48 })
49 except Exception as e:
50 logger.error("Failed to list skills: %s", e)
51 return jsonify({'success': False, 'error': str(e)}), 500
53 @app.route('/api/openclaw/skills/install', methods=['POST'])
54 def _openclaw_install_skill():
55 from flask import request, jsonify
56 data = request.get_json(silent=True) or {}
57 slug = data.get('slug', '')
58 if not slug:
59 return jsonify({'success': False, 'error': 'slug required'}), 400
61 try:
62 from integrations.openclaw.clawhub_adapter import (
63 install_skill, get_clawhub_provider
64 )
65 skill = install_skill(
66 slug,
67 version=data.get('version'),
68 force=data.get('force', False),
69 )
70 if skill:
71 get_clawhub_provider().invalidate()
72 return jsonify({
73 'success': True,
74 'skill': {
75 'name': skill.name,
76 'description': skill.description,
77 'version': skill.version,
78 },
79 })
80 return jsonify({'success': False, 'error': 'Install failed'}), 404
81 except Exception as e:
82 logger.error("Install failed: %s", e)
83 return jsonify({'success': False, 'error': str(e)}), 500
85 @app.route('/api/openclaw/skills/uninstall', methods=['POST'])
86 def _openclaw_uninstall_skill():
87 from flask import request, jsonify
88 data = request.get_json(silent=True) or {}
89 slug = data.get('slug', '')
90 if not slug:
91 return jsonify({'success': False, 'error': 'slug required'}), 400
93 try:
94 from integrations.openclaw.clawhub_adapter import (
95 uninstall_skill, get_clawhub_provider
96 )
97 removed = uninstall_skill(slug)
98 if removed:
99 get_clawhub_provider().invalidate()
100 return jsonify({'success': True, 'removed': removed})
101 except Exception as e:
102 return jsonify({'success': False, 'error': str(e)}), 500
104 @app.route('/api/openclaw/skills/search', methods=['GET'])
105 def _openclaw_search_skills():
106 from flask import request, jsonify
107 query = request.args.get('q', '')
108 if not query:
109 return jsonify({'success': False, 'error': 'q parameter required'}), 400
111 try:
112 from core.http_pool import pooled_get
113 except ImportError:
114 import requests
115 pooled_get = requests.get
117 try:
118 resp = pooled_get(
119 f"https://registry.clawhub.ai/api/skills/search",
120 params={'q': query, 'limit': 20},
121 timeout=10,
122 )
123 if hasattr(resp, 'json'):
124 data = resp.json()
125 else:
126 data = {'results': []}
127 return jsonify({'success': True, 'results': data.get('results', [])})
128 except Exception as e:
129 return jsonify({'success': False, 'error': str(e)}), 500
131 @app.route('/api/openclaw/status', methods=['GET'])
132 def _openclaw_status():
133 from flask import jsonify
134 try:
135 from integrations.openclaw.gateway_bridge import (
136 get_gateway_bridge, is_openclaw_installed, get_openclaw_version
137 )
138 bridge = get_gateway_bridge()
139 return jsonify({
140 'success': True,
141 'installed': is_openclaw_installed(),
142 'version': get_openclaw_version(),
143 'gateway': bridge.health(),
144 })
145 except Exception as e:
146 return jsonify({
147 'success': True,
148 'installed': False,
149 'version': None,
150 'gateway': {'connected': False},
151 })
153 @app.route('/api/openclaw/channels', methods=['GET'])
154 def _openclaw_channels():
155 from flask import jsonify
156 channels = [
157 {'id': 'whatsapp', 'name': 'WhatsApp', 'icon': 'chat'},
158 {'id': 'telegram', 'name': 'Telegram', 'icon': 'send'},
159 {'id': 'discord', 'name': 'Discord', 'icon': 'headset'},
160 {'id': 'slack', 'name': 'Slack', 'icon': 'tag'},
161 {'id': 'signal', 'name': 'Signal', 'icon': 'security'},
162 {'id': 'imessage', 'name': 'iMessage', 'icon': 'message'},
163 {'id': 'matrix', 'name': 'Matrix', 'icon': 'grid_on'},
164 {'id': 'teams', 'name': 'Teams', 'icon': 'groups'},
165 {'id': 'feishu', 'name': 'Feishu', 'icon': 'flight'},
166 {'id': 'line', 'name': 'LINE', 'icon': 'chat_bubble'},
167 ]
168 return jsonify({'success': True, 'channels': channels})
170 # ── Floating Assistant ─────────────────────────────────────
172 @app.route('/api/assistant/chat', methods=['POST'])
173 def _assistant_chat():
174 """Floating assistant chat — uses ALL HART capabilities.
176 This is the universal entry point: the floating chat bubble
177 can dispatch to any HART service (recipes, agents, TTS, vision,
178 OpenClaw skills, expert agents, etc.)
179 """
180 from flask import request, jsonify
181 data = request.get_json(silent=True) or {}
182 message = data.get('message', '')
183 user_id = data.get('user_id', '1')
184 context = data.get('context', {})
186 if not message:
187 return jsonify({'success': False, 'error': 'message required'}), 400
189 try:
190 # Route through the main HART chat pipeline
191 from core.http_pool import pooled_post
192 resp = pooled_post(
193 'http://localhost:6777/chat',
194 json={
195 'user_id': user_id,
196 'prompt_id': context.get('prompt_id', '99999'),
197 'prompt': message,
198 'create_agent': context.get('create_agent', False),
199 },
200 timeout=120,
201 )
202 if hasattr(resp, 'json'):
203 result = resp.json()
204 else:
205 result = {'response': str(resp)}
207 return jsonify({
208 'success': True,
209 'response': result.get('response', result.get('result', '')),
210 'source': 'hart_pipeline',
211 })
212 except Exception as e:
213 logger.error("Assistant chat error: %s", e)
214 return jsonify({'success': False, 'error': str(e)}), 500
216 @app.route('/api/assistant/capabilities', methods=['GET'])
217 def _assistant_capabilities():
218 """What the floating assistant can do."""
219 from flask import jsonify
220 capabilities = [
221 {
222 'id': 'chat',
223 'name': 'Chat',
224 'description': 'General conversation and task execution',
225 'icon': 'chat',
226 },
227 {
228 'id': 'recipe',
229 'name': 'Recipes',
230 'description': 'Execute trained recipes (fast, no LLM)',
231 'icon': 'receipt_long',
232 },
233 {
234 'id': 'agents',
235 'name': 'Agents',
236 'description': 'Create and manage autonomous agents',
237 'icon': 'smart_toy',
238 },
239 {
240 'id': 'vision',
241 'name': 'Vision',
242 'description': 'Analyze images and screenshots',
243 'icon': 'visibility',
244 },
245 {
246 'id': 'voice',
247 'name': 'Voice',
248 'description': 'Text-to-speech and voice cloning (offline)',
249 'icon': 'record_voice_over',
250 },
251 {
252 'id': 'expert',
253 'name': 'Expert Network',
254 'description': 'Query 96 specialized agents',
255 'icon': 'psychology',
256 },
257 {
258 'id': 'openclaw',
259 'name': 'OpenClaw Skills',
260 'description': '3,200+ ClawHub skills',
261 'icon': 'extension',
262 },
263 {
264 'id': 'code',
265 'name': 'Coding',
266 'description': 'Write, edit, and review code',
267 'icon': 'code',
268 },
269 {
270 'id': 'remote',
271 'name': 'Remote Desktop',
272 'description': 'Control remote devices',
273 'icon': 'desktop_windows',
274 },
275 {
276 'id': 'channels',
277 'name': 'Channels',
278 'description': 'Send messages via 30+ platforms',
279 'icon': 'forum',
280 },
281 ]
282 return jsonify({'success': True, 'capabilities': capabilities})
284 @app.route('/api/assistant/voice', methods=['POST'])
285 def _assistant_voice():
286 """Voice synthesis from the floating assistant."""
287 from flask import request, jsonify
288 data = request.get_json(silent=True) or {}
289 text = data.get('text', '')
290 voice = data.get('voice', 'alba')
292 if not text:
293 return jsonify({'success': False, 'error': 'text required'}), 400
295 try:
296 from integrations.audio.tts import get_tts_engine
297 engine = get_tts_engine()
298 path = engine.synthesize(text, voice=voice)
299 return jsonify({'success': True, 'audio_path': path})
300 except Exception as e:
301 return jsonify({'success': False, 'error': str(e)}), 500
303 logger.info("Registered OpenClaw + floating assistant API routes")