"""Sympathy-MCP server — VM operations as MCP tools.
Sympathetic magic: the connection between puppet and puppeteer.
Architecture:
- Command execution + file ops → SSH (transport.py)
- VM lifecycle (snapshot, restore, list, status) → Incus CLI (lifecycle.py)
- Host config → TOML file (config.py)
Run with: uv run python -m sympathy_mcp.server
Or via entry point: sympathy-mcp
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from mcp.server.fastmcp import FastMCP
from sympathy_mcp.transport import (
ExecResult,
vm_exec as _vm_exec,
vm_file_read as _vm_file_read,
vm_file_write as _vm_file_write,
vm_file_push as _vm_file_push,
vm_file_pull as _vm_file_pull,
)
from sympathy_mcp.lifecycle import (
VMInfo,
vm_snapshot as _vm_snapshot,
vm_restore as _vm_restore,
vm_status as _vm_status,
vm_list as _vm_list,
)
# ---------------------------------------------------------------------------
# Server setup
# ---------------------------------------------------------------------------
mcp = FastMCP(
"sympathy-mcp",
instructions=(
"Sympathy-MCP provides tools to operate remote VMs and containers. "
"Execute commands and transfer files via SSH, manage VM lifecycle "
"(snapshots, status) via Incus — all through structured tool calls "
"instead of freeform shell commands. "
"Hosts are configured in ~/.config/sympathy-mcp/hosts.toml."
),
)
# ---------------------------------------------------------------------------
# Helper: format results for MCP responses
# ---------------------------------------------------------------------------
def _format_exec_result(result: ExecResult) -> str:
"""Format an ExecResult as a readable string for MCP response."""
parts = []
if result.stdout:
parts.append(f"STDOUT:\n{result.stdout}")
if result.stderr:
parts.append(f"STDERR:\n{result.stderr}")
parts.append(f"EXIT CODE: {result.exit_code}")
return "\n".join(parts)
def _format_vm_info(info: VMInfo) -> str:
"""Format VMInfo as a readable string."""
lines = [
f"Name: {info.name}",
f"Status: {info.status}",
f"Type: {info.type}",
]
if info.architecture:
lines.append(f"Architecture: {info.architecture}")
if info.ipv4:
lines.append(f"IPv4: {info.ipv4}")
if info.ipv6:
lines.append(f"IPv6: {info.ipv6}")
if info.pid:
lines.append(f"PID: {info.pid}")
if info.processes:
lines.append(f"Processes: {info.processes}")
if info.memory_usage:
mb = info.memory_usage / (1024 * 1024)
lines.append(f"Memory: {mb:.1f} MB")
if info.cpu_usage:
secs = info.cpu_usage / 1_000_000_000
lines.append(f"CPU Time: {secs:.2f}s")
if info.snapshots:
lines.append(f"Snapshots: {', '.join(info.snapshots)}")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# MCP Tools — Command Execution & File Ops (via SSH)
# ---------------------------------------------------------------------------
@mcp.tool()
async def vm_exec(
vm: str,
command: str,
workdir: str = "/",
timeout: int = 120,
) -> str:
"""Execute a command on a remote host via SSH.
Runs the command via `sh -c` so shell features (pipes, redirects, etc.)
work as expected. Captures both stdout and stderr.
Args:
vm: Name of the host (as configured in hosts.toml).
command: Shell command to execute.
workdir: Working directory on the remote host (default: /).
timeout: Maximum seconds to wait (default: 120).
Returns:
Command output with stdout, stderr, and exit code.
"""
try:
result = await _vm_exec(vm, command, workdir=workdir, timeout=timeout)
return _format_exec_result(result)
except (ValueError, KeyError, TimeoutError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_file_read(vm: str, path: str) -> str:
"""Read a file from a remote host via SSH.
Args:
vm: Name of the host (as configured in hosts.toml).
path: Absolute path to the file on the remote host.
Returns:
File contents as a string, or an error message.
"""
try:
content = await _vm_file_read(vm, path)
return content
except (ValueError, KeyError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_file_write(vm: str, path: str, content: str) -> str:
"""Write content to a file on a remote host via SSH.
Handles arbitrary content safely by piping through SSH stdin.
Args:
vm: Name of the host (as configured in hosts.toml).
path: Absolute path where the file should be written.
content: The content to write to the file.
Returns:
Success confirmation or error message.
"""
try:
result = await _vm_file_write(vm, path, content)
if result.exit_code == 0:
return f"Successfully wrote {len(content)} bytes to {vm}:{path}"
return f"ERROR writing file: {result.stderr.strip()}"
except (ValueError, KeyError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_file_push(
vm: str,
local_path: str,
remote_path: str,
) -> str:
"""Push a file from the local host to a remote host via SCP.
Args:
vm: Name of the host (as configured in hosts.toml).
local_path: Path to the file on the local host.
remote_path: Absolute destination path on the remote host.
Returns:
Success confirmation or error message.
"""
try:
result = await _vm_file_push(vm, local_path, remote_path)
if result.exit_code == 0:
return f"Successfully pushed {local_path} -> {vm}:{remote_path}"
return f"ERROR pushing file: {result.stderr.strip()}"
except (ValueError, KeyError, FileNotFoundError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_file_pull(
vm: str,
remote_path: str,
local_path: str,
) -> str:
"""Pull a file from a remote host to the local host via SCP.
Args:
vm: Name of the host (as configured in hosts.toml).
remote_path: Absolute path to the file on the remote host.
local_path: Destination path on the local host.
Returns:
Success confirmation or error message.
"""
try:
result = await _vm_file_pull(vm, remote_path, local_path)
if result.exit_code == 0:
return f"Successfully pulled {vm}:{remote_path} -> {local_path}"
return f"ERROR pulling file: {result.stderr.strip()}"
except (ValueError, KeyError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
# ---------------------------------------------------------------------------
# MCP Tools — VM Lifecycle (via Incus CLI)
# ---------------------------------------------------------------------------
@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}"
@mcp.tool()
async def vm_restore(vm: str, snapshot: str) -> str:
"""Restore a VM/container to a named snapshot (Incus).
Args:
vm: Name of the VM or container.
snapshot: Name of the snapshot to restore.
Returns:
Success confirmation or error message.
"""
try:
result = await _vm_restore(vm, snapshot)
if result.exit_code == 0:
return f"Restored {vm} to snapshot '{snapshot}'"
return f"ERROR restoring snapshot: {result.stderr.strip()}"
except (ValueError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_status(vm: str) -> str:
"""Get the status of a VM/container (Incus).
Returns state (running/stopped), IP address, resource usage,
and available snapshots.
Args:
vm: Name of the VM or container.
Returns:
Formatted status information or error message.
"""
try:
info = await _vm_status(vm)
return _format_vm_info(info)
except (ValueError, RuntimeError, OSError) as e:
return f"ERROR: {e}"
@mcp.tool()
async def vm_list() -> str:
"""List all available VMs and containers with their status (Incus).
Returns:
Formatted list of all instances, or a message if none exist.
"""
try:
vms = await _vm_list()
if not vms:
return "No VMs or containers found."
lines = [f"{'Name':<20} {'Status':<12} {'Type':<18} {'IPv4':<16}"]
lines.append("-" * 66)
for vm in vms:
lines.append(
f"{vm.name:<20} {vm.status:<12} {vm.type:<18} {vm.ipv4:<16}"
)
return "\n".join(lines)
except (RuntimeError, OSError) as e:
return f"ERROR: {e}"
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
"""Run the Sympathy-MCP server via stdio transport."""
asyncio.run(mcp.run_stdio_async())
if __name__ == "__main__":
main()
