Partially prettier tool outputs
This commit is contained in:
17
src/proxmox_mcp/formatting/__init__.py
Normal file
17
src/proxmox_mcp/formatting/__init__.py
Normal file
@@ -0,0 +1,17 @@
|
||||
"""
|
||||
Proxmox MCP formatting package for styled output.
|
||||
"""
|
||||
|
||||
from .theme import ProxmoxTheme
|
||||
from .colors import ProxmoxColors
|
||||
from .formatters import ProxmoxFormatters
|
||||
from .templates import ProxmoxTemplates
|
||||
from .components import ProxmoxComponents
|
||||
|
||||
__all__ = [
|
||||
'ProxmoxTheme',
|
||||
'ProxmoxColors',
|
||||
'ProxmoxFormatters',
|
||||
'ProxmoxTemplates',
|
||||
'ProxmoxComponents'
|
||||
]
|
||||
116
src/proxmox_mcp/formatting/colors.py
Normal file
116
src/proxmox_mcp/formatting/colors.py
Normal file
@@ -0,0 +1,116 @@
|
||||
"""
|
||||
Color utilities for Proxmox MCP output styling.
|
||||
"""
|
||||
from typing import Optional
|
||||
from .theme import ProxmoxTheme
|
||||
|
||||
class ProxmoxColors:
|
||||
"""ANSI color definitions and utilities for terminal output."""
|
||||
|
||||
# Foreground colors
|
||||
BLACK = '\033[30m'
|
||||
RED = '\033[31m'
|
||||
GREEN = '\033[32m'
|
||||
YELLOW = '\033[33m'
|
||||
BLUE = '\033[34m'
|
||||
MAGENTA = '\033[35m'
|
||||
CYAN = '\033[36m'
|
||||
WHITE = '\033[37m'
|
||||
|
||||
# Background colors
|
||||
BG_BLACK = '\033[40m'
|
||||
BG_RED = '\033[41m'
|
||||
BG_GREEN = '\033[42m'
|
||||
BG_YELLOW = '\033[43m'
|
||||
BG_BLUE = '\033[44m'
|
||||
BG_MAGENTA = '\033[45m'
|
||||
BG_CYAN = '\033[46m'
|
||||
BG_WHITE = '\033[47m'
|
||||
|
||||
# Styles
|
||||
BOLD = '\033[1m'
|
||||
DIM = '\033[2m'
|
||||
ITALIC = '\033[3m'
|
||||
UNDERLINE = '\033[4m'
|
||||
BLINK = '\033[5m'
|
||||
REVERSE = '\033[7m'
|
||||
HIDDEN = '\033[8m'
|
||||
STRIKE = '\033[9m'
|
||||
|
||||
# Reset
|
||||
RESET = '\033[0m'
|
||||
|
||||
@classmethod
|
||||
def colorize(cls, text: str, color: str, style: Optional[str] = None) -> str:
|
||||
"""Add color and optional style to text with theme awareness.
|
||||
|
||||
Args:
|
||||
text: Text to colorize
|
||||
color: ANSI color code
|
||||
style: Optional ANSI style code
|
||||
|
||||
Returns:
|
||||
Formatted text string
|
||||
"""
|
||||
if not ProxmoxTheme.USE_COLORS:
|
||||
return text
|
||||
|
||||
if style:
|
||||
return f"{style}{color}{text}{cls.RESET}"
|
||||
return f"{color}{text}{cls.RESET}"
|
||||
|
||||
@classmethod
|
||||
def status_color(cls, status: str) -> str:
|
||||
"""Get appropriate color for a status value.
|
||||
|
||||
Args:
|
||||
status: Status string to get color for
|
||||
|
||||
Returns:
|
||||
ANSI color code
|
||||
"""
|
||||
status = status.lower()
|
||||
if status in ['online', 'running', 'success']:
|
||||
return cls.GREEN
|
||||
elif status in ['offline', 'stopped', 'error']:
|
||||
return cls.RED
|
||||
elif status in ['pending', 'warning']:
|
||||
return cls.YELLOW
|
||||
return cls.BLUE
|
||||
|
||||
@classmethod
|
||||
def resource_color(cls, resource_type: str) -> str:
|
||||
"""Get appropriate color for a resource type.
|
||||
|
||||
Args:
|
||||
resource_type: Resource type to get color for
|
||||
|
||||
Returns:
|
||||
ANSI color code
|
||||
"""
|
||||
resource_type = resource_type.lower()
|
||||
if resource_type in ['node', 'vm', 'container']:
|
||||
return cls.CYAN
|
||||
elif resource_type in ['cpu', 'memory', 'network']:
|
||||
return cls.YELLOW
|
||||
elif resource_type in ['storage', 'disk']:
|
||||
return cls.MAGENTA
|
||||
return cls.BLUE
|
||||
|
||||
@classmethod
|
||||
def metric_color(cls, value: float, warning: float = 80.0, critical: float = 90.0) -> str:
|
||||
"""Get appropriate color for a metric value based on thresholds.
|
||||
|
||||
Args:
|
||||
value: Metric value (typically percentage)
|
||||
warning: Warning threshold
|
||||
critical: Critical threshold
|
||||
|
||||
Returns:
|
||||
ANSI color code
|
||||
"""
|
||||
if value >= critical:
|
||||
return cls.RED
|
||||
elif value >= warning:
|
||||
return cls.YELLOW
|
||||
return cls.GREEN
|
||||
172
src/proxmox_mcp/formatting/components.py
Normal file
172
src/proxmox_mcp/formatting/components.py
Normal file
@@ -0,0 +1,172 @@
|
||||
"""
|
||||
Reusable UI components for Proxmox MCP output.
|
||||
"""
|
||||
from typing import List, Optional
|
||||
from .colors import ProxmoxColors
|
||||
from .theme import ProxmoxTheme
|
||||
|
||||
class ProxmoxComponents:
|
||||
"""Reusable UI components for formatted output."""
|
||||
|
||||
@staticmethod
|
||||
def create_table(headers: List[str], rows: List[List[str]], title: Optional[str] = None) -> str:
|
||||
"""Create an ASCII table with optional title.
|
||||
|
||||
Args:
|
||||
headers: List of column headers
|
||||
rows: List of row data
|
||||
title: Optional table title
|
||||
|
||||
Returns:
|
||||
Formatted table string
|
||||
"""
|
||||
# Calculate column widths considering multi-line content
|
||||
widths = [len(header) for header in headers]
|
||||
for row in rows:
|
||||
for i, cell in enumerate(row):
|
||||
cell_lines = str(cell).split('\n')
|
||||
max_line_length = max(len(line) for line in cell_lines)
|
||||
widths[i] = max(widths[i], max_line_length)
|
||||
|
||||
# Create separator line
|
||||
separator = "+" + "+".join("-" * (w + 2) for w in widths) + "+"
|
||||
|
||||
# Calculate total width for title
|
||||
total_width = sum(widths) + len(widths) + 1
|
||||
|
||||
# Build table
|
||||
result = []
|
||||
|
||||
# Add title if provided
|
||||
if title:
|
||||
# Center the title
|
||||
title_str = ProxmoxColors.colorize(title, ProxmoxColors.CYAN, ProxmoxColors.BOLD)
|
||||
padding = (total_width - len(title) - 2) // 2 # -2 for the border chars
|
||||
title_separator = "+" + "-" * (total_width - 2) + "+"
|
||||
result.extend([
|
||||
title_separator,
|
||||
"|" + " " * padding + title_str + " " * (total_width - padding - len(title) - 2) + "|",
|
||||
title_separator
|
||||
])
|
||||
|
||||
# Add headers
|
||||
header = "|" + "|".join(f" {ProxmoxColors.colorize(h, ProxmoxColors.CYAN):<{w}} " for w, h in zip(widths, headers)) + "|"
|
||||
result.extend([separator, header, separator])
|
||||
|
||||
# Add rows with multi-line cell support
|
||||
for row in rows:
|
||||
# Split each cell into lines
|
||||
cell_lines = [str(cell).split('\n') for cell in row]
|
||||
max_lines = max(len(lines) for lines in cell_lines)
|
||||
|
||||
# Pad cells with fewer lines
|
||||
padded_cells = []
|
||||
for lines in cell_lines:
|
||||
if len(lines) < max_lines:
|
||||
lines.extend([''] * (max_lines - len(lines)))
|
||||
padded_cells.append(lines)
|
||||
|
||||
# Create row strings for each line
|
||||
for line_idx in range(max_lines):
|
||||
line_parts = []
|
||||
for col_idx, cell_lines in enumerate(padded_cells):
|
||||
line = cell_lines[line_idx]
|
||||
line_parts.append(f" {line:<{widths[col_idx]}} ")
|
||||
result.append("|" + "|".join(line_parts) + "|")
|
||||
|
||||
# Add separator after each row except the last
|
||||
if row != rows[-1]:
|
||||
result.append(separator)
|
||||
|
||||
result.append(separator)
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def create_progress_bar(value: float, total: float, width: int = 20) -> str:
|
||||
"""Create a progress bar with percentage.
|
||||
|
||||
Args:
|
||||
value: Current value
|
||||
total: Maximum value
|
||||
width: Width of progress bar in characters
|
||||
|
||||
Returns:
|
||||
Formatted progress bar string
|
||||
"""
|
||||
percentage = min(100, (value / total * 100) if total > 0 else 0)
|
||||
filled = int(width * percentage / 100)
|
||||
color = ProxmoxColors.metric_color(percentage)
|
||||
|
||||
bar = "█" * filled + "░" * (width - filled)
|
||||
return f"{ProxmoxColors.colorize(bar, color)} {percentage:.1f}%"
|
||||
|
||||
@staticmethod
|
||||
def create_resource_usage(used: float, total: float, label: str, emoji: str) -> str:
|
||||
"""Create a resource usage display with progress bar.
|
||||
|
||||
Args:
|
||||
used: Used amount
|
||||
total: Total amount
|
||||
label: Resource label
|
||||
emoji: Resource emoji
|
||||
|
||||
Returns:
|
||||
Formatted resource usage string
|
||||
"""
|
||||
from .formatters import ProxmoxFormatters
|
||||
percentage = (used / total * 100) if total > 0 else 0
|
||||
progress = ProxmoxComponents.create_progress_bar(used, total)
|
||||
|
||||
return (
|
||||
f"{emoji} {label}:\n"
|
||||
f" {progress}\n"
|
||||
f" {ProxmoxFormatters.format_bytes(used)} / {ProxmoxFormatters.format_bytes(total)}"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create_key_value_grid(data: dict, columns: int = 2) -> str:
|
||||
"""Create a grid of key-value pairs.
|
||||
|
||||
Args:
|
||||
data: Dictionary of key-value pairs
|
||||
columns: Number of columns in grid
|
||||
|
||||
Returns:
|
||||
Formatted grid string
|
||||
"""
|
||||
# Calculate max widths for each column
|
||||
items = list(data.items())
|
||||
rows = [items[i:i + columns] for i in range(0, len(items), columns)]
|
||||
|
||||
key_widths = [0] * columns
|
||||
val_widths = [0] * columns
|
||||
|
||||
for row in rows:
|
||||
for i, (key, val) in enumerate(row):
|
||||
key_widths[i] = max(key_widths[i], len(str(key)))
|
||||
val_widths[i] = max(val_widths[i], len(str(val)))
|
||||
|
||||
# Format rows
|
||||
result = []
|
||||
for row in rows:
|
||||
formatted_items = []
|
||||
for i, (key, val) in enumerate(row):
|
||||
key_str = ProxmoxColors.colorize(f"{key}:", ProxmoxColors.CYAN)
|
||||
formatted_items.append(f"{key_str:<{key_widths[i] + 10}} {val:<{val_widths[i]}}")
|
||||
result.append(" ".join(formatted_items))
|
||||
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def create_status_badge(status: str) -> str:
|
||||
"""Create a status badge with emoji.
|
||||
|
||||
Args:
|
||||
status: Status string
|
||||
|
||||
Returns:
|
||||
Formatted status badge string
|
||||
"""
|
||||
status = status.lower()
|
||||
emoji = ProxmoxTheme.get_status_emoji(status)
|
||||
return f"{emoji} {status.upper()}"
|
||||
158
src/proxmox_mcp/formatting/formatters.py
Normal file
158
src/proxmox_mcp/formatting/formatters.py
Normal file
@@ -0,0 +1,158 @@
|
||||
"""
|
||||
Core formatting functions for Proxmox MCP output.
|
||||
"""
|
||||
from typing import List, Union, Dict, Any
|
||||
from .theme import ProxmoxTheme
|
||||
from .colors import ProxmoxColors
|
||||
|
||||
class ProxmoxFormatters:
|
||||
"""Core formatting functions for Proxmox data."""
|
||||
|
||||
@staticmethod
|
||||
def format_bytes(bytes_value: int) -> str:
|
||||
"""Format bytes with proper units.
|
||||
|
||||
Args:
|
||||
bytes_value: Number of bytes
|
||||
|
||||
Returns:
|
||||
Formatted string with appropriate unit
|
||||
"""
|
||||
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
|
||||
if bytes_value < 1024:
|
||||
return f"{bytes_value:.2f} {unit}"
|
||||
bytes_value /= 1024
|
||||
return f"{bytes_value:.2f} TB"
|
||||
|
||||
@staticmethod
|
||||
def format_uptime(seconds: int) -> str:
|
||||
"""Format uptime in seconds to human readable format.
|
||||
|
||||
Args:
|
||||
seconds: Uptime in seconds
|
||||
|
||||
Returns:
|
||||
Formatted uptime string
|
||||
"""
|
||||
days = seconds // 86400
|
||||
hours = (seconds % 86400) // 3600
|
||||
minutes = (seconds % 3600) // 60
|
||||
|
||||
parts = []
|
||||
if days > 0:
|
||||
parts.append(f"{days}d")
|
||||
if hours > 0:
|
||||
parts.append(f"{hours}h")
|
||||
if minutes > 0:
|
||||
parts.append(f"{minutes}m")
|
||||
|
||||
return f"{ProxmoxTheme.METRICS['uptime']} " + " ".join(parts) if parts else "0m"
|
||||
|
||||
@staticmethod
|
||||
def format_percentage(value: float, warning: float = 80.0, critical: float = 90.0) -> str:
|
||||
"""Format percentage with color based on thresholds.
|
||||
|
||||
Args:
|
||||
value: Percentage value
|
||||
warning: Warning threshold
|
||||
critical: Critical threshold
|
||||
|
||||
Returns:
|
||||
Formatted percentage string
|
||||
"""
|
||||
color = ProxmoxColors.metric_color(value, warning, critical)
|
||||
return ProxmoxColors.colorize(f"{value:.1f}%", color)
|
||||
|
||||
@staticmethod
|
||||
def format_status(status: str) -> str:
|
||||
"""Format status with emoji and color.
|
||||
|
||||
Args:
|
||||
status: Status string
|
||||
|
||||
Returns:
|
||||
Formatted status string
|
||||
"""
|
||||
status = status.lower()
|
||||
emoji = ProxmoxTheme.get_status_emoji(status)
|
||||
color = ProxmoxColors.status_color(status)
|
||||
return f"{emoji} {ProxmoxColors.colorize(status.upper(), color)}"
|
||||
|
||||
@staticmethod
|
||||
def format_resource_header(resource_type: str, name: str) -> str:
|
||||
"""Format resource header with emoji and styling.
|
||||
|
||||
Args:
|
||||
resource_type: Type of resource
|
||||
name: Resource name
|
||||
|
||||
Returns:
|
||||
Formatted header string
|
||||
"""
|
||||
emoji = ProxmoxTheme.get_resource_emoji(resource_type)
|
||||
color = ProxmoxColors.resource_color(resource_type)
|
||||
return f"\n{emoji} {ProxmoxColors.colorize(name, color, ProxmoxColors.BOLD)}"
|
||||
|
||||
@staticmethod
|
||||
def format_section_header(title: str, section_type: str = 'header') -> str:
|
||||
"""Format section header with emoji and border.
|
||||
|
||||
Args:
|
||||
title: Section title
|
||||
section_type: Type of section for emoji selection
|
||||
|
||||
Returns:
|
||||
Formatted section header
|
||||
"""
|
||||
emoji = ProxmoxTheme.get_section_emoji(section_type)
|
||||
header = f"{emoji} {title}"
|
||||
border = "═" * len(header)
|
||||
return f"\n{header}\n{border}\n"
|
||||
|
||||
@staticmethod
|
||||
def format_key_value(key: str, value: str, emoji: str = "") -> str:
|
||||
"""Format key-value pair with optional emoji.
|
||||
|
||||
Args:
|
||||
key: Label/key
|
||||
value: Value to display
|
||||
emoji: Optional emoji prefix
|
||||
|
||||
Returns:
|
||||
Formatted key-value string
|
||||
"""
|
||||
key_str = ProxmoxColors.colorize(key, ProxmoxColors.CYAN)
|
||||
prefix = f"{emoji} " if emoji else ""
|
||||
return f"{prefix}{key_str}: {value}"
|
||||
|
||||
@staticmethod
|
||||
def format_command_output(success: bool, command: str, output: str, error: str = None) -> str:
|
||||
"""Format command execution output.
|
||||
|
||||
Args:
|
||||
success: Whether command succeeded
|
||||
command: The command that was executed
|
||||
output: Command output
|
||||
error: Optional error message
|
||||
|
||||
Returns:
|
||||
Formatted command output string
|
||||
"""
|
||||
status_emoji = ProxmoxTheme.ACTIONS['success' if success else 'error']
|
||||
cmd_emoji = ProxmoxTheme.ACTIONS['command']
|
||||
|
||||
result = [
|
||||
ProxmoxFormatters.format_section_header("Console Command Result"),
|
||||
f"{status_emoji} Status: {ProxmoxColors.colorize('SUCCESS' if success else 'FAILED', ProxmoxColors.GREEN if success else ProxmoxColors.RED)}",
|
||||
f"{cmd_emoji} Command: {command}",
|
||||
"\n📤 Output:",
|
||||
output.strip()
|
||||
]
|
||||
|
||||
if error:
|
||||
result.extend([
|
||||
"\n❌ Error:",
|
||||
ProxmoxColors.colorize(error.strip(), ProxmoxColors.RED)
|
||||
])
|
||||
|
||||
return "\n".join(result)
|
||||
186
src/proxmox_mcp/formatting/templates.py
Normal file
186
src/proxmox_mcp/formatting/templates.py
Normal file
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
Output templates for Proxmox MCP resource types.
|
||||
"""
|
||||
from typing import Dict, List, Any
|
||||
from .formatters import ProxmoxFormatters
|
||||
from .theme import ProxmoxTheme
|
||||
from .colors import ProxmoxColors
|
||||
from .components import ProxmoxComponents
|
||||
|
||||
class ProxmoxTemplates:
|
||||
"""Output templates for different Proxmox resource types."""
|
||||
|
||||
@staticmethod
|
||||
def node_list(nodes: List[Dict[str, Any]]) -> str:
|
||||
"""Template for node list output.
|
||||
|
||||
Args:
|
||||
nodes: List of node data dictionaries
|
||||
|
||||
Returns:
|
||||
Formatted node list string
|
||||
"""
|
||||
result = [f"{ProxmoxTheme.RESOURCES['node']} Proxmox Nodes"]
|
||||
|
||||
for node in nodes:
|
||||
# Get node status
|
||||
status = node.get("status", "unknown")
|
||||
|
||||
# Get memory info
|
||||
memory = node.get("memory", {})
|
||||
memory_used = memory.get("used", 0)
|
||||
memory_total = memory.get("total", 0)
|
||||
memory_percent = (memory_used / memory_total * 100) if memory_total > 0 else 0
|
||||
|
||||
# Format node info
|
||||
result.extend([
|
||||
"", # Empty line between nodes
|
||||
f"{ProxmoxTheme.RESOURCES['node']} {node['node']}",
|
||||
f" • Status: {status.upper()}",
|
||||
f" • Uptime: {ProxmoxFormatters.format_uptime(node.get('uptime', 0))}",
|
||||
f" • CPU Cores: {node.get('maxcpu', 'N/A')}",
|
||||
f" • Memory: {ProxmoxFormatters.format_bytes(memory_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(memory_total)} ({memory_percent:.1f}%)"
|
||||
])
|
||||
|
||||
# Add disk usage if available
|
||||
disk = node.get("disk", {})
|
||||
if disk:
|
||||
disk_used = disk.get("used", 0)
|
||||
disk_total = disk.get("total", 0)
|
||||
disk_percent = (disk_used / disk_total * 100) if disk_total > 0 else 0
|
||||
result.append(
|
||||
f" • Disk: {ProxmoxFormatters.format_bytes(disk_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(disk_total)} ({disk_percent:.1f}%)"
|
||||
)
|
||||
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def node_status(node: str, status: Dict[str, Any]) -> str:
|
||||
"""Template for detailed node status output.
|
||||
|
||||
Args:
|
||||
node: Node name
|
||||
status: Node status data
|
||||
|
||||
Returns:
|
||||
Formatted node status string
|
||||
"""
|
||||
memory = status.get("memory", {})
|
||||
memory_used = memory.get("used", 0)
|
||||
memory_total = memory.get("total", 0)
|
||||
memory_percent = (memory_used / memory_total * 100) if memory_total > 0 else 0
|
||||
|
||||
result = [
|
||||
f"{ProxmoxTheme.RESOURCES['node']} Node: {node}",
|
||||
f" • Status: {status.get('status', 'unknown').upper()}",
|
||||
f" • Uptime: {ProxmoxFormatters.format_uptime(status.get('uptime', 0))}",
|
||||
f" • CPU Cores: {status.get('maxcpu', 'N/A')}",
|
||||
f" • Memory: {ProxmoxFormatters.format_bytes(memory_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(memory_total)} ({memory_percent:.1f}%)"
|
||||
]
|
||||
|
||||
# Add disk usage if available
|
||||
disk = status.get("disk", {})
|
||||
if disk:
|
||||
disk_used = disk.get("used", 0)
|
||||
disk_total = disk.get("total", 0)
|
||||
disk_percent = (disk_used / disk_total * 100) if disk_total > 0 else 0
|
||||
result.append(
|
||||
f" • Disk: {ProxmoxFormatters.format_bytes(disk_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(disk_total)} ({disk_percent:.1f}%)"
|
||||
)
|
||||
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def vm_list(vms: List[Dict[str, Any]]) -> str:
|
||||
"""Template for VM list output.
|
||||
|
||||
Args:
|
||||
vms: List of VM data dictionaries
|
||||
|
||||
Returns:
|
||||
Formatted VM list string
|
||||
"""
|
||||
result = [f"{ProxmoxTheme.RESOURCES['vm']} Virtual Machines"]
|
||||
|
||||
for vm in vms:
|
||||
memory = vm.get("memory", {})
|
||||
memory_used = memory.get("used", 0)
|
||||
memory_total = memory.get("total", 0)
|
||||
memory_percent = (memory_used / memory_total * 100) if memory_total > 0 else 0
|
||||
|
||||
result.extend([
|
||||
"", # Empty line between VMs
|
||||
f"{ProxmoxTheme.RESOURCES['vm']} {vm['name']} (ID: {vm['vmid']})",
|
||||
f" • Status: {vm['status'].upper()}",
|
||||
f" • Node: {vm['node']}",
|
||||
f" • CPU Cores: {vm.get('cpus', 'N/A')}",
|
||||
f" • Memory: {ProxmoxFormatters.format_bytes(memory_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(memory_total)} ({memory_percent:.1f}%)"
|
||||
])
|
||||
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def storage_list(storage: List[Dict[str, Any]]) -> str:
|
||||
"""Template for storage list output.
|
||||
|
||||
Args:
|
||||
storage: List of storage data dictionaries
|
||||
|
||||
Returns:
|
||||
Formatted storage list string
|
||||
"""
|
||||
result = [f"{ProxmoxTheme.RESOURCES['storage']} Storage Pools"]
|
||||
|
||||
for store in storage:
|
||||
used = store.get("used", 0)
|
||||
total = store.get("total", 0)
|
||||
percent = (used / total * 100) if total > 0 else 0
|
||||
|
||||
result.extend([
|
||||
"", # Empty line between storage pools
|
||||
f"{ProxmoxTheme.RESOURCES['storage']} {store['storage']}",
|
||||
f" • Status: {store.get('status', 'unknown').upper()}",
|
||||
f" • Type: {store['type']}",
|
||||
f" • Usage: {ProxmoxFormatters.format_bytes(used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(total)} ({percent:.1f}%)"
|
||||
])
|
||||
|
||||
return "\n".join(result)
|
||||
|
||||
@staticmethod
|
||||
def container_list(containers: List[Dict[str, Any]]) -> str:
|
||||
"""Template for container list output.
|
||||
|
||||
Args:
|
||||
containers: List of container data dictionaries
|
||||
|
||||
Returns:
|
||||
Formatted container list string
|
||||
"""
|
||||
if not containers:
|
||||
return f"{ProxmoxTheme.RESOURCES['container']} No containers found"
|
||||
|
||||
result = [f"{ProxmoxTheme.RESOURCES['container']} Containers"]
|
||||
|
||||
for container in containers:
|
||||
memory = container.get("memory", {})
|
||||
memory_used = memory.get("used", 0)
|
||||
memory_total = memory.get("total", 0)
|
||||
memory_percent = (memory_used / memory_total * 100) if memory_total > 0 else 0
|
||||
|
||||
result.extend([
|
||||
"", # Empty line between containers
|
||||
f"{ProxmoxTheme.RESOURCES['container']} {container['name']} (ID: {container['vmid']})",
|
||||
f" • Status: {container['status'].upper()}",
|
||||
f" • Node: {container['node']}",
|
||||
f" • CPU Cores: {container.get('cpus', 'N/A')}",
|
||||
f" • Memory: {ProxmoxFormatters.format_bytes(memory_used)} / "
|
||||
f"{ProxmoxFormatters.format_bytes(memory_total)} ({memory_percent:.1f}%)"
|
||||
])
|
||||
|
||||
return "\n".join(result)
|
||||
102
src/proxmox_mcp/formatting/theme.py
Normal file
102
src/proxmox_mcp/formatting/theme.py
Normal file
@@ -0,0 +1,102 @@
|
||||
"""
|
||||
Theme configuration for Proxmox MCP output styling.
|
||||
"""
|
||||
|
||||
class ProxmoxTheme:
|
||||
"""Theme configuration for Proxmox MCP output."""
|
||||
|
||||
# Feature flags
|
||||
USE_EMOJI = True
|
||||
USE_COLORS = True
|
||||
|
||||
# Status indicators with emojis
|
||||
STATUS = {
|
||||
'online': '🟢',
|
||||
'offline': '🔴',
|
||||
'running': '▶️',
|
||||
'stopped': '⏹️',
|
||||
'unknown': '❓',
|
||||
'pending': '⏳',
|
||||
'error': '❌',
|
||||
'warning': '⚠️',
|
||||
}
|
||||
|
||||
# Resource type indicators
|
||||
RESOURCES = {
|
||||
'node': '🖥️',
|
||||
'vm': '🗃️',
|
||||
'container': '📦',
|
||||
'storage': '💾',
|
||||
'cpu': '⚡',
|
||||
'memory': '🧠',
|
||||
'network': '🌐',
|
||||
'disk': '💿',
|
||||
'backup': '📼',
|
||||
'snapshot': '📸',
|
||||
'template': '📋',
|
||||
'pool': '🏊',
|
||||
}
|
||||
|
||||
# Action and operation indicators
|
||||
ACTIONS = {
|
||||
'success': '✅',
|
||||
'error': '❌',
|
||||
'warning': '⚠️',
|
||||
'info': 'ℹ️',
|
||||
'command': '🔧',
|
||||
'start': '▶️',
|
||||
'stop': '⏹️',
|
||||
'restart': '🔄',
|
||||
'delete': '🗑️',
|
||||
'edit': '✏️',
|
||||
'create': '➕',
|
||||
'migrate': '➡️',
|
||||
'clone': '📑',
|
||||
'lock': '🔒',
|
||||
'unlock': '🔓',
|
||||
}
|
||||
|
||||
# Section and grouping indicators
|
||||
SECTIONS = {
|
||||
'header': '📌',
|
||||
'details': '📝',
|
||||
'statistics': '📊',
|
||||
'configuration': '⚙️',
|
||||
'logs': '📜',
|
||||
'tasks': '📋',
|
||||
'users': '👥',
|
||||
'permissions': '🔑',
|
||||
}
|
||||
|
||||
# Measurement and metric indicators
|
||||
METRICS = {
|
||||
'percentage': '%',
|
||||
'temperature': '🌡️',
|
||||
'uptime': '⏳',
|
||||
'bandwidth': '📶',
|
||||
'latency': '⚡',
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_status_emoji(cls, status: str) -> str:
|
||||
"""Get emoji for a status value with fallback."""
|
||||
status = status.lower()
|
||||
return cls.STATUS.get(status, cls.STATUS['unknown'])
|
||||
|
||||
@classmethod
|
||||
def get_resource_emoji(cls, resource: str) -> str:
|
||||
"""Get emoji for a resource type with fallback."""
|
||||
resource = resource.lower()
|
||||
return cls.RESOURCES.get(resource, '📦')
|
||||
|
||||
@classmethod
|
||||
def get_action_emoji(cls, action: str) -> str:
|
||||
"""Get emoji for an action with fallback."""
|
||||
action = action.lower()
|
||||
return cls.ACTIONS.get(action, cls.ACTIONS['info'])
|
||||
|
||||
@classmethod
|
||||
def get_section_emoji(cls, section: str) -> str:
|
||||
"""Get emoji for a section type with fallback."""
|
||||
section = section.lower()
|
||||
return cls.SECTIONS.get(section, cls.SECTIONS['details'])
|
||||
@@ -2,9 +2,10 @@
|
||||
Base classes and utilities for Proxmox MCP tools.
|
||||
"""
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from mcp.types import TextContent as Content
|
||||
from proxmoxer import ProxmoxAPI
|
||||
from ..formatting import ProxmoxTemplates
|
||||
|
||||
class ProxmoxTool:
|
||||
"""Base class for Proxmox MCP tools."""
|
||||
@@ -18,17 +19,36 @@ class ProxmoxTool:
|
||||
self.proxmox = proxmox_api
|
||||
self.logger = logging.getLogger(f"proxmox-mcp.{self.__class__.__name__.lower()}")
|
||||
|
||||
def _format_response(self, data: Any) -> List[Content]:
|
||||
"""Format response data into MCP content.
|
||||
def _format_response(self, data: Any, resource_type: Optional[str] = None) -> List[Content]:
|
||||
"""Format response data into MCP content using templates.
|
||||
|
||||
Args:
|
||||
data: Data to format
|
||||
resource_type: Optional type of resource for template selection
|
||||
|
||||
Returns:
|
||||
List of Content objects
|
||||
"""
|
||||
import json
|
||||
return [Content(type="text", text=json.dumps(data, indent=2))]
|
||||
if resource_type == "nodes":
|
||||
formatted = ProxmoxTemplates.node_list(data)
|
||||
elif resource_type == "node_status":
|
||||
# For node_status, data should be a tuple of (node_name, status_dict)
|
||||
if isinstance(data, tuple) and len(data) == 2:
|
||||
formatted = ProxmoxTemplates.node_status(data[0], data[1])
|
||||
else:
|
||||
formatted = ProxmoxTemplates.node_status("unknown", data)
|
||||
elif resource_type == "vms":
|
||||
formatted = ProxmoxTemplates.vm_list(data)
|
||||
elif resource_type == "storage":
|
||||
formatted = ProxmoxTemplates.storage_list(data)
|
||||
elif resource_type == "containers":
|
||||
formatted = ProxmoxTemplates.container_list(data)
|
||||
else:
|
||||
# Fallback to JSON formatting for unknown types
|
||||
import json
|
||||
formatted = json.dumps(data, indent=2)
|
||||
|
||||
return [Content(type="text", text=formatted)]
|
||||
|
||||
def _handle_error(self, operation: str, error: Exception) -> None:
|
||||
"""Handle and log errors.
|
||||
|
||||
@@ -20,8 +20,37 @@ class NodeTools(ProxmoxTool):
|
||||
"""
|
||||
try:
|
||||
result = self.proxmox.nodes.get()
|
||||
nodes = [{"node": node["node"], "status": node["status"]} for node in result]
|
||||
return self._format_response(nodes)
|
||||
nodes = []
|
||||
|
||||
# Get detailed info for each node
|
||||
for node in result:
|
||||
node_name = node["node"]
|
||||
try:
|
||||
# Get detailed status for each node
|
||||
status = self.proxmox.nodes(node_name).status.get()
|
||||
nodes.append({
|
||||
"node": node_name,
|
||||
"status": node["status"],
|
||||
"uptime": status.get("uptime", 0),
|
||||
"maxcpu": status.get("cpuinfo", {}).get("cpus", "N/A"),
|
||||
"memory": {
|
||||
"used": status.get("memory", {}).get("used", 0),
|
||||
"total": status.get("memory", {}).get("total", 0)
|
||||
}
|
||||
})
|
||||
except Exception:
|
||||
# Fallback to basic info if detailed status fails
|
||||
nodes.append({
|
||||
"node": node_name,
|
||||
"status": node["status"],
|
||||
"uptime": 0,
|
||||
"maxcpu": "N/A",
|
||||
"memory": {
|
||||
"used": node.get("maxmem", 0) - node.get("mem", 0),
|
||||
"total": node.get("maxmem", 0)
|
||||
}
|
||||
})
|
||||
return self._format_response(nodes, "nodes")
|
||||
except Exception as e:
|
||||
self._handle_error("get nodes", e)
|
||||
|
||||
@@ -40,6 +69,6 @@ class NodeTools(ProxmoxTool):
|
||||
"""
|
||||
try:
|
||||
result = self.proxmox.nodes(node).status.get()
|
||||
return self._format_response(result)
|
||||
return self._format_response((node, result), "node_status")
|
||||
except Exception as e:
|
||||
self._handle_error(f"get status for node {node}", e)
|
||||
|
||||
@@ -20,12 +20,33 @@ class StorageTools(ProxmoxTool):
|
||||
"""
|
||||
try:
|
||||
result = self.proxmox.storage.get()
|
||||
storage = [{
|
||||
"storage": storage["storage"],
|
||||
"type": storage["type"],
|
||||
"content": storage.get("content", []),
|
||||
"enabled": storage.get("enabled", True)
|
||||
} for storage in result]
|
||||
return self._format_response(storage)
|
||||
storage = []
|
||||
|
||||
for store in result:
|
||||
# Get detailed storage info including usage
|
||||
try:
|
||||
status = self.proxmox.nodes(store.get("node", "localhost")).storage(store["storage"]).status.get()
|
||||
storage.append({
|
||||
"storage": store["storage"],
|
||||
"type": store["type"],
|
||||
"content": store.get("content", []),
|
||||
"status": "online" if store.get("enabled", True) else "offline",
|
||||
"used": status.get("used", 0),
|
||||
"total": status.get("total", 0),
|
||||
"available": status.get("avail", 0)
|
||||
})
|
||||
except Exception:
|
||||
# If detailed status fails, add basic info
|
||||
storage.append({
|
||||
"storage": store["storage"],
|
||||
"type": store["type"],
|
||||
"content": store.get("content", []),
|
||||
"status": "online" if store.get("enabled", True) else "offline",
|
||||
"used": 0,
|
||||
"total": 0,
|
||||
"available": 0
|
||||
})
|
||||
|
||||
return self._format_response(storage, "storage")
|
||||
except Exception as e:
|
||||
self._handle_error("get storage", e)
|
||||
|
||||
@@ -36,9 +36,14 @@ class VMTools(ProxmoxTool):
|
||||
"vmid": vm["vmid"],
|
||||
"name": vm["name"],
|
||||
"status": vm["status"],
|
||||
"node": node["node"]
|
||||
"node": node["node"],
|
||||
"cpu": vm.get("cpu", 0),
|
||||
"memory": {
|
||||
"used": vm.get("mem", 0),
|
||||
"total": vm.get("maxmem", 0)
|
||||
}
|
||||
} for vm in vms])
|
||||
return self._format_response(result)
|
||||
return self._format_response(result, "vms")
|
||||
except Exception as e:
|
||||
self._handle_error("get VMs", e)
|
||||
|
||||
@@ -59,6 +64,14 @@ class VMTools(ProxmoxTool):
|
||||
"""
|
||||
try:
|
||||
result = await self.console_manager.execute_command(node, vmid, command)
|
||||
return self._format_response(result)
|
||||
# Use the command output formatter from ProxmoxFormatters
|
||||
from ..formatting import ProxmoxFormatters
|
||||
formatted = ProxmoxFormatters.format_command_output(
|
||||
success=result["success"],
|
||||
command=command,
|
||||
output=result["output"],
|
||||
error=result.get("error")
|
||||
)
|
||||
return [Content(type="text", text=formatted)]
|
||||
except Exception as e:
|
||||
self._handle_error(f"execute command on VM {vmid}", e)
|
||||
|
||||
Reference in New Issue
Block a user