Coverage for integrations / remote_desktop / agent_tools.py: 32.9%
158 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Remote Desktop Agent Tools — AutoGen tool definitions for remote desktop.
4Agents can programmatically:
5 - Offer remote help (start hosting, share device ID + password)
6 - Request screen view (connect view-only, take screenshot)
7 - Execute remote actions (click, type, key via connected session)
8 - Transfer files between devices
9 - Manage sessions
11Follows core/agent_tools.py pattern:
12 build_remote_desktop_tools(ctx) -> List[(name, desc, func)]
13 register_remote_desktop_tools(tools, helper, executor)
14"""
15import logging
16from typing import Annotated, Any, List, Optional, Tuple
18logger = logging.getLogger('hevolve.remote_desktop')
21def build_remote_desktop_tools(ctx) -> List[Tuple[str, str, Any]]:
22 """Build remote desktop tool closures for AutoGen agents.
24 Args:
25 ctx: dict with session variables (user_id, prompt_id, etc.)
27 Returns:
28 List of (name, description, func) tuples.
29 """
30 user_id = ctx.get('user_id', 'agent')
31 tools: List[Tuple[str, str, Any]] = []
33 # ── offer_remote_help ─────────────────────────────────────
35 def offer_remote_help(
36 allow_control: Annotated[bool, "Allow remote control (False=view-only)"] = True,
37 ) -> str:
38 """Start hosting this device for remote desktop. Returns Device ID + password."""
39 try:
40 from integrations.remote_desktop.orchestrator import get_orchestrator
41 result = get_orchestrator().start_hosting(
42 allow_control=allow_control,
43 user_id=user_id,
44 )
45 if result.get('status') == 'error':
46 return f"Failed to start hosting: {result.get('error')}"
47 return (
48 f"Remote desktop hosting started.\n"
49 f"Device ID: {result.get('formatted_id', 'Unknown')}\n"
50 f"Password: {result.get('password', 'N/A')}\n"
51 f"Mode: {result.get('mode', 'full_control')}\n"
52 f"Engine: {result.get('engine', 'auto')}\n"
53 f"Share Device ID + Password with the viewer to connect."
54 )
55 except Exception as e:
56 return f"Failed to start hosting: {e}"
58 tools.append((
59 "offer_remote_help",
60 "Start hosting this device for remote desktop access. Returns Device ID and password to share.",
61 offer_remote_help,
62 ))
64 # ── request_screen_view ───────────────────────────────────
66 def request_screen_view(
67 device_id: Annotated[str, "Remote device ID to connect to"],
68 password: Annotated[str, "Access password for the remote device"],
69 ) -> str:
70 """Connect to a remote device in view-only mode."""
71 try:
72 from integrations.remote_desktop.orchestrator import get_orchestrator
73 result = get_orchestrator().connect(
74 device_id=device_id,
75 password=password,
76 mode='view_only',
77 gui=False,
78 user_id=user_id,
79 )
80 if result.get('status') == 'connected':
81 return (f"Connected to {device_id} (view-only) "
82 f"via {result.get('engine', 'auto')}")
83 return f"Connection failed: {result.get('error', 'Unknown error')}"
84 except Exception as e:
85 return f"Connection error: {e}"
87 tools.append((
88 "request_screen_view",
89 "Connect to a remote device in view-only mode to observe the screen.",
90 request_screen_view,
91 ))
93 # ── remote_execute_action ─────────────────────────────────
95 def remote_execute_action(
96 action: Annotated[str, "Action type: click, type, key, hotkey, scroll"],
97 x: Annotated[Optional[int], "X coordinate (for click/move)"] = None,
98 y: Annotated[Optional[int], "Y coordinate (for click/move)"] = None,
99 text: Annotated[Optional[str], "Text to type or key name"] = None,
100 ) -> str:
101 """Execute a mouse/keyboard action on the connected remote device."""
102 try:
103 from integrations.remote_desktop.input_handler import InputHandler
104 handler = InputHandler()
106 event = {'type': action}
107 if x is not None:
108 event['x'] = x
109 if y is not None:
110 event['y'] = y
111 if text is not None:
112 if action in ('type',):
113 event['text'] = text
114 elif action in ('key', 'hotkey'):
115 event['key'] = text
117 result = handler.handle_input_event(event)
118 return f"Action executed: {result}"
119 except Exception as e:
120 return f"Action failed: {e}"
122 tools.append((
123 "remote_execute_action",
124 "Execute a click, type, key, hotkey, or scroll action on the connected remote device.",
125 remote_execute_action,
126 ))
128 # ── remote_screenshot ─────────────────────────────────────
130 def remote_screenshot() -> str:
131 """Capture a screenshot of the local or connected remote screen."""
132 try:
133 from integrations.remote_desktop.frame_capture import FrameCapture
134 capture = FrameCapture()
135 frame = capture.capture_frame()
136 if frame:
137 return f"Screenshot captured ({len(frame)} bytes JPEG)"
138 return "Screenshot capture failed — no frame returned"
139 except Exception as e:
140 return f"Screenshot failed: {e}"
142 tools.append((
143 "remote_screenshot",
144 "Capture a screenshot of the current screen (local or remote session).",
145 remote_screenshot,
146 ))
148 # ── remote_transfer_file ──────────────────────────────────
150 def remote_transfer_file(
151 device_id: Annotated[str, "Target device ID"],
152 local_path: Annotated[str, "Local file path to transfer"],
153 ) -> str:
154 """Transfer a file to a connected remote device via RustDesk."""
155 try:
156 import os
157 if not os.path.exists(local_path):
158 return f"File not found: {local_path}"
160 from integrations.remote_desktop.rustdesk_bridge import get_rustdesk_bridge
161 bridge = get_rustdesk_bridge()
162 if not bridge.available:
163 return "RustDesk not installed (required for file transfer)"
165 ok, msg = bridge.connect(device_id, file_transfer=True)
166 if ok:
167 return f"File transfer session opened to {device_id}: {msg}"
168 return f"File transfer failed: {msg}"
169 except Exception as e:
170 return f"File transfer error: {e}"
172 tools.append((
173 "remote_transfer_file",
174 "Transfer a file to a remote device using RustDesk file transfer.",
175 remote_transfer_file,
176 ))
178 # ── get_remote_sessions ───────────────────────────────────
180 def get_remote_sessions() -> str:
181 """List all active remote desktop sessions."""
182 try:
183 from integrations.remote_desktop.session_manager import get_session_manager
184 sm = get_session_manager()
185 sessions = sm.get_active_sessions()
186 if not sessions:
187 return "No active remote desktop sessions."
188 lines = [f"Active sessions ({len(sessions)}):"]
189 for s in sessions:
190 lines.append(
191 f" {s.session_id[:8]} host={s.host_device_id[:12]} "
192 f"mode={s.mode.value} state={s.state.value} "
193 f"viewers={len(s.viewer_device_ids)}"
194 )
195 return "\n".join(lines)
196 except Exception as e:
197 return f"Error listing sessions: {e}"
199 tools.append((
200 "get_remote_sessions",
201 "List all active remote desktop sessions with their status.",
202 get_remote_sessions,
203 ))
205 # ── disconnect_remote ─────────────────────────────────────
207 def disconnect_remote(
208 session_id: Annotated[Optional[str], "Session ID to disconnect (all if empty)"] = None,
209 ) -> str:
210 """End a remote desktop session or all sessions."""
211 try:
212 from integrations.remote_desktop.session_manager import get_session_manager
213 sm = get_session_manager()
214 if session_id:
215 sm.disconnect_session(session_id)
216 return f"Disconnected session {session_id[:8]}"
217 else:
218 sessions = sm.get_active_sessions()
219 for s in sessions:
220 sm.disconnect_session(s.session_id)
221 return f"Disconnected {len(sessions)} session(s)"
222 except Exception as e:
223 return f"Disconnect error: {e}"
225 tools.append((
226 "disconnect_remote",
227 "Disconnect a specific remote desktop session or all sessions.",
228 disconnect_remote,
229 ))
231 # ── list_remote_windows ──────────────────────────────────
233 def list_remote_windows() -> str:
234 """List available application windows on this host for per-window streaming."""
235 try:
236 from integrations.remote_desktop.orchestrator import get_orchestrator
237 windows = get_orchestrator().list_remote_windows()
238 if not windows:
239 return "No application windows found."
240 lines = [f"Available windows ({len(windows)}):"]
241 for w in windows:
242 lines.append(
243 f" hwnd={w.get('hwnd')} \"{w.get('title', 'Untitled')}\" "
244 f"({w.get('process_name', 'unknown')})"
245 )
246 return "\n".join(lines)
247 except Exception as e:
248 return f"Error listing windows: {e}"
250 tools.append((
251 "list_remote_windows",
252 "List available application windows on the host for per-window streaming (tab detach).",
253 list_remote_windows,
254 ))
256 # ── stream_remote_window ──────────────────────────────────
258 def stream_remote_window(
259 window_title: Annotated[str, "Window title or pattern to stream"],
260 ) -> str:
261 """Start streaming a specific application window from the host."""
262 try:
263 from integrations.remote_desktop.window_capture import WindowEnumerator
264 enum = WindowEnumerator()
265 winfo = enum.get_window_by_title(window_title)
266 if not winfo:
267 return f"No window matching '{window_title}' found."
269 from integrations.remote_desktop.orchestrator import get_orchestrator
270 result = get_orchestrator().stream_window(
271 window_hwnd=winfo.hwnd,
272 window_title=winfo.title,
273 )
274 if result.get('status') == 'error':
275 return f"Failed: {result.get('error')}"
276 return (
277 f"Streaming window: {result.get('window_title')}\n"
278 f"Session: {result.get('session_id')}\n"
279 f"Process: {result.get('process_name', 'unknown')}"
280 )
281 except Exception as e:
282 return f"Error streaming window: {e}"
284 tools.append((
285 "stream_remote_window",
286 "Start streaming a specific application window from the host (tab detach).",
287 stream_remote_window,
288 ))
290 # ── list_peripherals ───────────────────────────────────────
292 def list_peripherals(
293 types: Annotated[Optional[str], "Filter by type: usb,bluetooth,gamepad (comma-sep, or None for all)"] = None,
294 ) -> str:
295 """List local peripheral devices available for forwarding to remote host."""
296 try:
297 from integrations.remote_desktop.orchestrator import get_orchestrator
298 type_list = [t.strip() for t in types.split(',')] if types else None
299 peripherals = get_orchestrator().list_peripherals(types=type_list)
300 if not peripherals:
301 return "No peripherals found."
302 lines = [f"Peripherals ({len(peripherals)}):"]
303 for p in peripherals:
304 fwd = " [FORWARDING]" if p.get('forwarded') else ""
305 lines.append(
306 f" {p.get('peripheral_id')} {p.get('name')} "
307 f"type={p.get('type')}{fwd}"
308 )
309 return "\n".join(lines)
310 except Exception as e:
311 return f"Error listing peripherals: {e}"
313 tools.append((
314 "list_peripherals",
315 "List local peripheral devices (USB, Bluetooth, gamepad) available for remote forwarding.",
316 list_peripherals,
317 ))
319 # ── forward_peripheral ─────────────────────────────────────
321 def forward_peripheral(
322 peripheral_id: Annotated[str, "Peripheral ID to forward"],
323 session_id: Annotated[Optional[str], "Remote session ID (auto-detect if empty)"] = None,
324 ) -> str:
325 """Forward a local peripheral device to the connected remote host."""
326 try:
327 from integrations.remote_desktop.orchestrator import get_orchestrator
328 orch = get_orchestrator()
330 # Auto-detect session if not provided
331 if not session_id:
332 sessions = orch.get_status().get('active_sessions', [])
333 if sessions:
334 session_id = sessions[0].get('session_id', '')
335 else:
336 return "No active session. Connect to a remote host first."
338 result = orch.forward_peripheral(session_id, peripheral_id)
339 if result.get('success'):
340 return (
341 f"Forwarding {result.get('type', 'device')}: {result.get('name')}\n"
342 f"Peripheral ID: {result.get('peripheral_id')}"
343 )
344 return f"Forward failed: {result.get('error', 'Unknown error')}"
345 except Exception as e:
346 return f"Forward error: {e}"
348 tools.append((
349 "forward_peripheral",
350 "Forward a local peripheral (USB, Bluetooth, gamepad) to the connected remote host.",
351 forward_peripheral,
352 ))
354 # ── discover_cast_targets ──────────────────────────────────
356 def discover_cast_targets() -> str:
357 """Discover DLNA/UPnP renderers (smart TVs, speakers) on the local network."""
358 try:
359 from integrations.remote_desktop.orchestrator import get_orchestrator
360 targets = get_orchestrator().discover_cast_targets()
361 if not targets:
362 return "No DLNA/UPnP renderers found on the network."
363 lines = [f"Cast targets ({len(targets)}):"]
364 for t in targets:
365 lines.append(
366 f" {t.get('device_id', 'unknown')[:12]} "
367 f"\"{t.get('friendly_name', 'Unknown')}\" "
368 f"at {t.get('ip', '?')}:{t.get('port', '?')}"
369 )
370 return "\n".join(lines)
371 except Exception as e:
372 return f"Error discovering cast targets: {e}"
374 tools.append((
375 "discover_cast_targets",
376 "Discover DLNA/UPnP renderers (smart TVs, speakers) for screen casting.",
377 discover_cast_targets,
378 ))
380 # ── cast_to_tv ─────────────────────────────────────────────
382 def cast_to_tv(
383 renderer_id: Annotated[str, "DLNA renderer device ID to cast to"],
384 session_id: Annotated[Optional[str], "Session to cast (auto-detect if empty)"] = None,
385 ) -> str:
386 """Cast a remote desktop session to a DLNA/UPnP renderer (smart TV)."""
387 try:
388 from integrations.remote_desktop.orchestrator import get_orchestrator
389 orch = get_orchestrator()
391 if not session_id:
392 sessions = orch.get_status().get('active_sessions', [])
393 if sessions:
394 session_id = sessions[0].get('session_id', '')
395 else:
396 return "No active session to cast."
398 result = orch.cast_to_device(session_id, renderer_id)
399 if result.get('success'):
400 return (
401 f"Casting to: {result.get('renderer_name', 'Unknown')}\n"
402 f"Stream URL: {result.get('stream_url', 'N/A')}\n"
403 f"Cast session: {result.get('cast_session_id', 'N/A')}"
404 )
405 return f"Cast failed: {result.get('error', 'Unknown error')}"
406 except Exception as e:
407 return f"Cast error: {e}"
409 tools.append((
410 "cast_to_tv",
411 "Cast a remote desktop session to a smart TV or DLNA/UPnP renderer.",
412 cast_to_tv,
413 ))
415 return tools
418def register_remote_desktop_tools(tools, helper, executor):
419 """Register remote desktop tools on an AutoGen helper/executor pair.
421 Same pattern as core/agent_tools.py:register_core_tools().
422 """
423 for name, desc, func in tools:
424 helper.register_for_llm(name=name, description=desc)(func)
425 executor.register_for_execution(name=name)(func)