Sympathy-MCP

Instructions

Implementation Reference

• src/sympathy_mcp/lifecycle.py:123-130 (handler)

  Core implementation of vm_snapshot that validates inputs and executes the 'incus snapshot create' command
  async def vm_snapshot(vm: str, name: str) -> ExecResult: """Create a snapshot of a VM/container. Uses `incus snapshot create <vm> <name>`. """ _validate_vm_name(vm) _validate_snapshot_name(name) return await _run_incus("snapshot", "create", vm, name)
• src/sympathy_mcp/server.py:224-241 (registration)

  MCP tool registration with @mcp.tool() decorator that wraps the implementation and provides user-friendly output messages
  @mcp.tool() async def vm_snapshot(vm: str, name: str) -> str: """Create a snapshot of a VM/container (Incus). Args: vm: Name of the VM or container. name: Name for the snapshot (alphanumeric, hyphens, underscores). Returns: Success confirmation or error message. """ try: result = await _vm_snapshot(vm, name) if result.exit_code == 0: return f"Snapshot '{name}' created for {vm}" return f"ERROR creating snapshot: {result.stderr.strip()}" except (ValueError, RuntimeError, OSError) as e: return f"ERROR: {e}"
• src/sympathy_mcp/lifecycle.py:54-73 (schema)

  Validation functions for VM names and snapshot names using regex patterns to ensure safe input values
  def _validate_vm_name(name: str) -> None: """Validate that a VM/container name is safe.""" if not name or not name.strip(): raise ValueError("VM name cannot be empty") if not _VALID_NAME.match(name): raise ValueError( f"Invalid VM name '{name}': must start with a letter, " "contain only alphanumeric characters and hyphens" ) def _validate_snapshot_name(name: str) -> None: """Validate a snapshot name.""" if not name or not name.strip(): raise ValueError("Snapshot name cannot be empty") if not re.match(r"^[a-zA-Z0-9_\-]+$", name): raise ValueError( f"Invalid snapshot name '{name}': " "only alphanumeric, hyphens, and underscores allowed" )
• src/sympathy_mcp/lifecycle.py:80-116 (helper)

  Helper function that executes incus CLI commands asynchronously, captures stdout/stderr, and returns an ExecResult with timeout handling
  async def _run_incus( *args: str, timeout: int = 120, ) -> ExecResult: """Run an incus CLI command and capture output. Args: *args: Arguments to pass to `incus`. timeout: Maximum seconds to wait. Returns: ExecResult with stdout, stderr, and exit code. """ proc = await asyncio.create_subprocess_exec( "incus", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.DEVNULL, ) try: stdout_bytes, stderr_bytes = await asyncio.wait_for( proc.communicate(), timeout=timeout, ) except asyncio.TimeoutError: proc.kill() await proc.wait() raise TimeoutError( f"incus command timed out after {timeout}s: incus {' '.join(args)}" ) return ExecResult( stdout=stdout_bytes.decode("utf-8", errors="replace"), stderr=stderr_bytes.decode("utf-8", errors="replace"), exit_code=proc.returncode or 0, )
• src/sympathy_mcp/lifecycle.py:21-28 (helper)

  Dataclass that represents the result of executing a CLI command with stdout, stderr, and exit_code fields
  @dataclass class ExecResult: """Result of running a CLI command.""" stdout: str stderr: str exit_code: int