# Directory Structure
```
├── .gitignore
├── docker-compose.yml
├── Dockerfile
├── LICENSE
├── proxmox-config
│ └── config.example.json
├── pyproject.toml
├── README.md
├── requirements-dev.in
├── requirements.in
├── setup.py
├── src
│ └── proxmox_mcp
│ ├── __init__.py
│ ├── config
│ │ ├── __init__.py
│ │ ├── loader.py
│ │ └── models.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── logging.py
│ │ └── proxmox.py
│ ├── formatting
│ │ ├── __init__.py
│ │ ├── colors.py
│ │ ├── components.py
│ │ ├── formatters.py
│ │ ├── templates.py
│ │ └── theme.py
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── cluster.py
│ │ ├── console
│ │ │ ├── __init__.py
│ │ │ └── manager.py
│ │ ├── containers.py
│ │ ├── definitions.py
│ │ ├── node.py
│ │ ├── storage.py
│ │ └── vm.py
│ └── utils
│ ├── __init__.py
│ ├── auth.py
│ └── logging.py
├── start_openapi.sh
├── start_server.sh
├── test_scripts
│ ├── README.md
│ ├── test_common.py
│ ├── test_create_vm.py
│ ├── test_openapi.py
│ ├── test_vm_power.py
│ └── test_vm_start.py
└── tests
├── __init__.py
├── test_server.py
└── test_vm_console.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
.env
.venv
env/
venv/
ENV/
# IDE
.idea/
.vscode/
*.swp
*.swo
.project
.pydevproject
.settings/
# Logs
*.log
logs/
# Test coverage
.coverage
htmlcov/
.tox/
.nox/
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# UV
.uv/
# Local configuration
config/config.json
proxmox-config/config.json
**/config.json
```
--------------------------------------------------------------------------------
/test_scripts/README.md:
--------------------------------------------------------------------------------
```markdown
# ProxmoxMCP Test Scripts
This folder contains various test scripts and demo programs for the ProxmoxMCP project.
## 📁 File Description
### 🔧 VM Management Tests
- **`test_vm_power.py`** - VM power management functionality test
- Test VM start, stop, restart and other operations
- Check VM status and available operations
- **`test_vm_start.py`** - VM startup functionality specific test
- Dedicated test for VM startup functionality
- Suitable for single VM startup testing
- **`test_create_vm.py`** - VM creation functionality test
- Test complete workflow of creating new virtual machines
- Verify 1 CPU core + 2GB RAM + 10GB storage configuration
### 🌐 API Tests
- **`test_openapi.py`** - OpenAPI service comprehensive test
- Test all API endpoints
- Include VM creation, power management and other functionalities
- Verify integration with Open WebUI
## 🚀 Usage
### Environment Setup
```bash
# Activate virtual environment
source ../.venv/bin/activate
# Set configuration path (if needed)
export PROXMOX_MCP_CONFIG=../proxmox-config/config.json
```
### Running Tests
#### 1. Test VM Power Management
```bash
python test_vm_power.py
```
#### 2. Test VM Creation
```bash
python test_create_vm.py
```
#### 3. Test OpenAPI Service
```bash
python test_openapi.py
```
#### 4. Test VM Startup
```bash
python test_vm_start.py
```
## 📋 Test Coverage
### ✅ Tested Features
- [x] VM list retrieval
- [x] VM status query
- [x] VM power management (start/stop/restart/shutdown)
- [x] VM creation (support custom CPU/memory/storage)
- [x] Storage type auto-detection
- [x] Disk format intelligent selection
- [x] OpenAPI service integration
- [x] Error handling verification
### 🎯 Test Scenarios
- **Basic functionality**: Connection, authentication, basic operations
- **VM lifecycle**: Create, start, stop, delete
- **Storage compatibility**: LVM, filesystem storage
- **API integration**: REST API calls and responses
- **Error recovery**: Exception handling
## 🔗 Related Documentation
- **Main project documentation**: [../README.md](../README.md)
- **VM creation guide**: [../VM_CREATION_GUIDE.md](../VM_CREATION_GUIDE.md)
- **OpenAPI deployment**: [../OPENAPI_DEPLOYMENT.md](../OPENAPI_DEPLOYMENT.md)
- **Quick deployment**: [../QUICK_DEPLOY_8811.md](../QUICK_DEPLOY_8811.md)
## 📊 Test Results Examples
### Success Cases
```
✅ VM 995: Created successfully (local-lvm, raw)
✅ VM 996: Created successfully (vm-storage, raw)
✅ VM 998: Created successfully (local-lvm, raw)
✅ VM 999: Created successfully (local-lvm, raw)
```
### API Endpoint Verification
```
✅ get_nodes: 200 - 134 chars
✅ get_vms: 200 - 1843 chars
✅ create_vm: 200 - VM created successfully
✅ start_vm: 200 - VM started successfully
```
## 🛠️ Troubleshooting
If tests fail, please check:
1. **Configuration file**: Whether `../proxmox-config/config.json` is correct
2. **Network connection**: Whether Proxmox server is reachable
3. **Authentication info**: Whether API token is valid
4. **Service status**: Whether OpenAPI service is running on port 8811
## 📝 Contributing Guidelines
When adding new tests, please:
1. Use descriptive filenames (e.g., `test_function_name.py`)
2. Include detailed docstrings
3. Add appropriate error handling
4. Update this README file
---
**Last Updated**: December 2024
**Maintainer**: ProxmoxMCP Development Team
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# ProxmoxMCP-Plus - Enhanced Proxmox MCP Server
An enhanced Python-based Model Context Protocol (MCP) server for interacting with Proxmox virtualization platforms. This project is built upon **[canvrno/ProxmoxMCP](https://github.com/canvrno/ProxmoxMCP)** with numerous new features and improvements, providing complete OpenAPI integration and more powerful virtualization management capabilities.
## Acknowledgments
This project is built upon the excellent open-source project [ProxmoxMCP](https://github.com/canvrno/ProxmoxMCP) by [@canvrno](https://github.com/canvrno). Thanks to the original author for providing the foundational framework and creative inspiration!
## 🆕 New Features and Improvements
### Major enhancements compared to the original version:
- ✨ **Complete VM Lifecycle Management**
- Brand new `create_vm` tool - Support for creating virtual machines with custom configurations
- New `delete_vm` tool - Safe VM deletion (with force deletion option)
- Enhanced intelligent storage type detection (LVM/file-based)
- 🔧 **Extended Power Management Features**
- `start_vm` - Start virtual machines
- `stop_vm` - Force stop virtual machines
- `shutdown_vm` - Graceful shutdown
- `reset_vm` - Restart virtual machines
- 🐳 **New Container Support**
- `get_containers` - List all LXC containers and their status
- `start_container` - Start LXC container
- `stop_container` - Stop LXC container
- `restart_container` - Restart LXC container (forcefully/gracefully)
- `update_container_resources` - Adjust container CPU, memory, swap, or extend disk
- 📊 **Enhanced Monitoring and Display**
- Improved storage pool status monitoring
- More detailed cluster health status checks
- Rich output formatting and themes
- 🌐 **Complete OpenAPI Integration**
- 11 complete REST API endpoints
- Production-ready Docker deployment
- Perfect Open WebUI integration
- Natural language VM creation support
- 🛡️ **Production-grade Security and Stability**
- Enhanced error handling mechanisms
- Comprehensive parameter validation
- Production-level logging
- Complete unit test coverage
## Built With
- [Cline](https://github.com/cline/cline) - Autonomous coding agent - Go faster with Cline
- [Proxmoxer](https://github.com/proxmoxer/proxmoxer) - Python wrapper for Proxmox API
- [MCP SDK](https://github.com/modelcontextprotocol/sdk) - Model Context Protocol SDK
- [Pydantic](https://docs.pydantic.dev/) - Data validation using Python type annotations
## Features
- 🤖 Full integration with Cline and Open WebUI
- 🛠️ Built with the official MCP SDK
- 🔒 Secure token-based authentication with Proxmox
- 🖥️ Complete VM lifecycle management (create, start, stop, reset, shutdown, delete)
- 💻 VM console command execution
- 🐳 LXC container management support
- 🗃️ Intelligent storage type detection (LVM/file-based)
- 📝 Configurable logging system
- ✅ Type-safe implementation with Pydantic
- 🎨 Rich output formatting with customizable themes
- 🌐 OpenAPI REST endpoints for integration
- 📡 11 fully functional API endpoints
## Installation
### Prerequisites
- UV package manager (recommended)
- Python 3.9 or higher
- Git
- Access to a Proxmox server with API token credentials
Before starting, ensure you have:
- [ ] Proxmox server hostname or IP
- [ ] Proxmox API token (see [API Token Setup](#proxmox-api-token-setup))
- [ ] UV installed (`pip install uv`)
### Option 1: Quick Install (Recommended)
1. Clone and set up environment:
```bash
# Clone repository
git clone https://github.com/RekklesNA/ProxmoxMCP-Plus.git
cd ProxmoxMCP-Plus
# Create and activate virtual environment
uv venv
source .venv/bin/activate # Linux/macOS
# OR
.\.venv\Scripts\Activate.ps1 # Windows
```
2. Install dependencies:
```bash
# Install with development dependencies
uv pip install -e ".[dev]"
```
3. Create configuration:
```bash
# Create config directory and copy template
mkdir -p proxmox-config
cp proxmox-config/config.example.json proxmox-config/config.json
```
4. Edit `proxmox-config/config.json`:
```json
{
"proxmox": {
"host": "PROXMOX_HOST", # Required: Your Proxmox server address
"port": 8006, # Optional: Default is 8006
"verify_ssl": false, # Optional: Set false for self-signed certs
"service": "PVE" # Optional: Default is PVE
},
"auth": {
"user": "USER@pve", # Required: Your Proxmox username
"token_name": "TOKEN_NAME", # Required: API token ID
"token_value": "TOKEN_VALUE" # Required: API token value
},
"logging": {
"level": "INFO", # Optional: DEBUG for more detail
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"file": "proxmox_mcp.log" # Optional: Log to file
}
}
```
### Verifying Installation
1. Check Python environment:
```bash
python -c "import proxmox_mcp; print('Installation OK')"
```
2. Run the tests:
```bash
pytest
```
3. Verify configuration:
```bash
# Linux/macOS
PROXMOX_MCP_CONFIG="proxmox-config/config.json" python -m proxmox_mcp.server
# Windows (PowerShell)
$env:PROXMOX_MCP_CONFIG="proxmox-config\config.json"; python -m proxmox_mcp.server
```
## Configuration
### Proxmox API Token Setup
1. Log into your Proxmox web interface
2. Navigate to Datacenter -> Permissions -> API Tokens
3. Create a new API token:
- Select a user (e.g., root@pam)
- Enter a token ID (e.g., "mcp-token")
- Uncheck "Privilege Separation" if you want full access
- Save and copy both the token ID and secret
## Running the Server
### Development Mode
For testing and development:
```bash
# Activate virtual environment first
source .venv/bin/activate # Linux/macOS
# OR
.\.venv\Scripts\Activate.ps1 # Windows
# Run the server
python -m proxmox_mcp.server
```
### OpenAPI Deployment (Production Ready)
Deploy ProxmoxMCP Plus as standard OpenAPI REST endpoints for integration with Open WebUI and other applications.
#### Quick OpenAPI Start
```bash
# Install mcpo (MCP-to-OpenAPI proxy)
pip install mcpo
# Start OpenAPI service on port 8811
./start_openapi.sh
```
#### Docker Deployment
```bash
# Build and run with Docker
docker build -t proxmox-mcp-api .
docker run -d --name proxmox-mcp-api -p 8811:8811 \
-v $(pwd)/proxmox-config:/app/proxmox-config proxmox-mcp-api
# Or use Docker Compose
docker-compose up -d
```
#### Access OpenAPI Service
Once deployed, access your service at:
- **📖 API Documentation**: http://your-server:8811/docs
- **🔧 OpenAPI Specification**: http://your-server:8811/openapi.json
- **❤️ Health Check**: http://your-server:8811/health
### Cline Desktop Integration
For Cline users, add this configuration to your MCP settings file (typically at `~/.config/Code/User/globalStorage/saoudrizwan.claude-dev/settings/cline_mcp_settings.json`):
```json
{
"mcpServers": {
"ProxmoxMCP-Plus": {
"command": "/absolute/path/to/ProxmoxMCP-Plus/.venv/bin/python",
"args": ["-m", "proxmox_mcp.server"],
"cwd": "/absolute/path/to/ProxmoxMCP-Plus",
"env": {
"PYTHONPATH": "/absolute/path/to/ProxmoxMCP-Plus/src",
"PROXMOX_MCP_CONFIG": "/absolute/path/to/ProxmoxMCP-Plus/proxmox-config/config.json",
"PROXMOX_HOST": "your-proxmox-host",
"PROXMOX_USER": "username@pve",
"PROXMOX_TOKEN_NAME": "token-name",
"PROXMOX_TOKEN_VALUE": "token-value",
"PROXMOX_PORT": "8006",
"PROXMOX_VERIFY_SSL": "false",
"PROXMOX_SERVICE": "PVE",
"LOG_LEVEL": "DEBUG"
},
"disabled": false,
"autoApprove": []
}
}
}
```
## Available Tools & API Endpoints
The server provides 11 comprehensive MCP tools and corresponding REST API endpoints:
### VM Management Tools
#### create_vm
Create a new virtual machine with specified resources.
**Parameters:**
- `node` (string, required): Name of the node
- `vmid` (string, required): ID for the new VM
- `name` (string, required): Name for the VM
- `cpus` (integer, required): Number of CPU cores (1-32)
- `memory` (integer, required): Memory in MB (512-131072)
- `disk_size` (integer, required): Disk size in GB (5-1000)
- `storage` (string, optional): Storage pool name
- `ostype` (string, optional): OS type (default: l26)
**API Endpoint:**
```http
POST /create_vm
Content-Type: application/json
{
"node": "pve",
"vmid": "200",
"name": "my-vm",
"cpus": 1,
"memory": 2048,
"disk_size": 10
}
```
**Example Response:**
```
🎉 VM 200 created successfully!
📋 VM Configuration:
• Name: my-vm
• Node: pve
• VM ID: 200
• CPU Cores: 1
• Memory: 2048 MB (2.0 GB)
• Disk: 10 GB (local-lvm, raw format)
• Storage Type: lvmthin
• Network: virtio (bridge=vmbr0)
• QEMU Agent: Enabled
🔧 Task ID: UPID:pve:001AB729:0442E853:682FF380:qmcreate:200:root@pam!mcp
```
#### VM Power Management 🆕
**start_vm**: Start a virtual machine
```http
POST /start_vm
{"node": "pve", "vmid": "200"}
```
**stop_vm**: Force stop a virtual machine
```http
POST /stop_vm
{"node": "pve", "vmid": "200"}
```
**shutdown_vm**: Gracefully shutdown a virtual machine
```http
POST /shutdown_vm
{"node": "pve", "vmid": "200"}
```
**reset_vm**: Reset (restart) a virtual machine
```http
POST /reset_vm
{"node": "pve", "vmid": "200"}
```
**delete_vm** 🆕: Completely delete a virtual machine
```http
POST /delete_vm
{"node": "pve", "vmid": "200", "force": false}
```
### 🆕 Container Management Tools
#### get_containers 🆕
List all LXC containers across the cluster.
**API Endpoint:** `POST /get_containers`
**Example Response:**
```
🐳 Containers
🐳 nginx-server (ID: 200)
• Status: RUNNING
• Node: pve
• CPU Cores: 2
• Memory: 1.5 GB / 2.0 GB (75.0%)
```
### Monitoring Tools
#### get_nodes
Lists all nodes in the Proxmox cluster.
**API Endpoint:** `POST /get_nodes`
**Example Response:**
```
🖥️ Proxmox Nodes
🖥️ pve-compute-01
• Status: ONLINE
• Uptime: ⏳ 156d 12h
• CPU Cores: 64
• Memory: 186.5 GB / 512.0 GB (36.4%)
```
#### get_node_status
Get detailed status of a specific node.
**Parameters:**
- `node` (string, required): Name of the node
**API Endpoint:** `POST /get_node_status`
#### get_vms
List all VMs across the cluster.
**API Endpoint:** `POST /get_vms`
#### get_storage
List available storage pools.
**API Endpoint:** `POST /get_storage`
#### get_cluster_status
Get overall cluster status and health.
**API Endpoint:** `POST /get_cluster_status`
#### execute_vm_command
Execute a command in a VM's console using QEMU Guest Agent.
**Parameters:**
- `node` (string, required): Name of the node where VM is running
- `vmid` (string, required): ID of the VM
- `command` (string, required): Command to execute
**API Endpoint:** `POST /execute_vm_command`
**Requirements:**
- VM must be running
- QEMU Guest Agent must be installed and running in the VM
## Open WebUI Integration
### Configure Open WebUI
1. Access your Open WebUI instance
2. Navigate to **Settings** → **Connections** → **OpenAPI**
3. Add new API configuration:
```json
{
"name": "Proxmox MCP API Plus",
"base_url": "http://your-server:8811",
"api_key": "",
"description": "Enhanced Proxmox Virtualization Management API"
}
```
### Natural Language VM Creation
Users can now request VMs using natural language:
- **"Can you create a VM with 1 cpu core and 2 GB ram with 10GB of storage disk"**
- **"Create a new VM for testing with minimal resources"**
- **"I need a development server with 4 cores and 8GB RAM"**
The AI assistant will automatically call the appropriate APIs and provide detailed feedback.
## Storage Type Support
### Intelligent Storage Detection
ProxmoxMCP Plus automatically detects storage types and selects appropriate disk formats:
#### LVM Storage (local-lvm, vm-storage)
- ✅ Format: `raw`
- ✅ High performance
- ⚠️ No cloud-init image support
#### File-based Storage (local, NFS, CIFS)
- ✅ Format: `qcow2`
- ✅ Cloud-init support
- ✅ Flexible snapshot capabilities
## Project Structure
```
ProxmoxMCP-Plus/
├── 📁 src/ # Source code
│ └── proxmox_mcp/
│ ├── server.py # Main MCP server implementation
│ ├── config/ # Configuration handling
│ ├── core/ # Core functionality
│ ├── formatting/ # Output formatting and themes
│ ├── tools/ # Tool implementations
│ │ ├── vm.py # VM management (create/power) 🆕
│ │ ├── container.py # Container management 🆕
│ │ └── console/ # VM console operations
│ └── utils/ # Utilities (auth, logging)
│
├── 📁 tests/ # Unit test suite
├── 📁 test_scripts/ # Integration tests & demos
│ ├── README.md # Test documentation
│ ├── test_vm_power.py # VM power management tests 🆕
│ ├── test_vm_start.py # VM startup tests
│ ├── test_create_vm.py # VM creation tests 🆕
│ └── test_openapi.py # OpenAPI service tests
│
├── 📁 proxmox-config/ # Configuration files
│ └── config.json # Server configuration
│
├── 📄 Configuration Files
│ ├── pyproject.toml # Project metadata
│ ├── docker-compose.yml # Docker orchestration
│ ├── Dockerfile # Docker image definition
│ └── requirements.in # Dependencies
│
├── 📄 Scripts
│ ├── start_server.sh # MCP server launcher
│ └── start_openapi.sh # OpenAPI service launcher
│
└── 📄 Documentation
├── README.md # This file
├── VM_CREATION_GUIDE.md # VM creation guide
├── OPENAPI_DEPLOYMENT.md # OpenAPI deployment
└── LICENSE # MIT License
```
## Testing
### Run Unit Tests
```bash
pytest
```
### Run Integration Tests
```bash
cd test_scripts
# Test VM power management
python test_vm_power.py
# Test VM creation
python test_create_vm.py
# Test OpenAPI service
python test_openapi.py
```
### API Testing with curl
```bash
# Test node listing
curl -X POST "http://your-server:8811/get_nodes" \
-H "Content-Type: application/json" \
-d "{}"
# Test VM creation
curl -X POST "http://your-server:8811/create_vm" \
-H "Content-Type: application/json" \
-d '{
"node": "pve",
"vmid": "300",
"name": "test-vm",
"cpus": 1,
"memory": 2048,
"disk_size": 10
}'
```
## Production Security
### API Key Authentication
Set up secure API access:
```bash
export PROXMOX_API_KEY="your-secure-api-key"
export PROXMOX_MCP_CONFIG="/app/proxmox-config/config.json"
```
### Nginx Reverse Proxy
Example nginx configuration:
```nginx
server {
listen 80;
server_name your-domain.com;
location / {
proxy_pass http://localhost:8811;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
```
## Troubleshooting
### Common Issues
1. **Port already in use**
```bash
netstat -tlnp | grep 8811
# Change port if needed
mcpo --port 8812 -- ./start_server.sh
```
2. **Configuration errors**
```bash
# Verify config file
cat proxmox-config/config.json
```
3. **Connection issues**
```bash
# Test Proxmox connectivity
curl -k https://your-proxmox:8006/api2/json/version
```
### View Logs
```bash
# View service logs
tail -f proxmox_mcp.log
# Docker logs
docker logs proxmox-mcp-api -f
```
## Deployment Status
### ✅ Feature Completion: 100%
- [x] VM Creation (user requirement: 1 CPU + 2GB RAM + 10GB storage) 🆕
- [x] VM Power Management (start VPN-Server ID:101) 🆕
- [x] VM Deletion Feature 🆕
- [x] Container Management (LXC) 🆕
- [x] Storage Compatibility (LVM/file-based)
- [x] OpenAPI Integration (port 8811)
- [x] Open WebUI Integration
- [x] Error Handling & Validation
- [x] Complete Documentation & Testing
### Production Ready!
**ProxmoxMCP Plus is now fully ready for production use!**
When users say **"Can you create a VM with 1 cpu core and 2 GB ram with 10GB of storage disk"**, the AI assistant can:
1. 📞 Call the `create_vm` API
2. 🔧 Automatically select appropriate storage and format
3. 🎯 Create VMs that match requirements
4. 📊 Return detailed configuration information
5. 💡 Provide next-step recommendations
## Development
After activating your virtual environment:
- Run tests: `pytest`
- Format code: `black .`
- Type checking: `mypy .`
- Lint: `ruff .`
## License
MIT License
## Special Thanks
- Thanks to [@canvrno](https://github.com/canvrno) for the excellent foundational project [ProxmoxMCP](https://github.com/canvrno/ProxmoxMCP)
- Thanks to the Proxmox community for providing the powerful virtualization platform
- Thanks to all contributors and users for their support
---
**Ready to Deploy!** 🎉 Your enhanced Proxmox MCP service with OpenAPI integration is ready for production use.
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/config/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/core/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Test suite for the Proxmox MCP server.
"""
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/__init__.py:
--------------------------------------------------------------------------------
```python
"""
MCP tools for interacting with Proxmox hypervisors.
"""
__all__ = []
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/utils/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Utility functions and helpers for the Proxmox MCP server.
"""
__all__ = []
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/console/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Console management package for Proxmox MCP.
"""
from .manager import VMConsoleManager
__all__ = ['VMConsoleManager']
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Proxmox MCP Server - A Model Context Protocol server for interacting with Proxmox hypervisors.
"""
from .server import ProxmoxMCPServer
__version__ = "0.1.0"
__all__ = ["ProxmoxMCPServer"]
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/__init__.py:
--------------------------------------------------------------------------------
```python
"""
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'
]
```
--------------------------------------------------------------------------------
/proxmox-config/config.example.json:
--------------------------------------------------------------------------------
```json
{
"proxmox": {
"host": "your-proxmox-host-ip",
"port": 8006,
"verify_ssl": false,
"service": "PVE"
},
"auth": {
"user": "username@pve",
"token_name": "your-token-name",
"token_value": "your-token-value"
},
"logging": {
"level": "DEBUG",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"file": "proxmox_mcp.log"
}
}
```
--------------------------------------------------------------------------------
/docker-compose.yml:
--------------------------------------------------------------------------------
```yaml
version: '3.8'
services:
proxmox-mcp-api:
build: .
ports:
- "8811:8811"
volumes:
- ./proxmox-config:/app/proxmox-config:ro
environment:
- PROXMOX_MCP_CONFIG=/app/proxmox-config/config.json
- API_HOST=0.0.0.0
- API_PORT=8811
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8811/health"]
interval: 30s
timeout: 10s
retries: 3
networks:
- proxmox-network
networks:
proxmox-network:
driver: bridge
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use Python 3.11 slim image as base
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
git \
curl \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
RUN pip install mcpo uv
# Copy project files
COPY . .
# Create virtual environment and install dependencies
RUN uv venv && \
. .venv/bin/activate && \
uv pip install -e ".[dev]"
# Expose port
EXPOSE 8811
# Set environment variables
ENV PROXMOX_MCP_CONFIG="/app/proxmox-config/config.json"
ENV API_HOST="0.0.0.0"
ENV API_PORT="8811"
# Startup command
CMD ["mcpo", "--host", "0.0.0.0", "--port", "8811", "--", \
"/bin/bash", "-c", "cd /app && source .venv/bin/activate && python -m proxmox_mcp.server"]
```
--------------------------------------------------------------------------------
/start_server.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
"""
Proxmox MCP server startup script
"""
echo "🚀 Starting Proxmox MCP server..."
echo ""
# Check virtual environment
if [ ! -d ".venv" ]; then
echo "❌ Virtual environment does not exist, please run installation steps first"
exit 1
fi
# Activate virtual environment
source .venv/bin/activate
# Set environment variables
export PROXMOX_MCP_CONFIG="proxmox-config/config.json"
# Check configuration file
if [ ! -f "$PROXMOX_MCP_CONFIG" ]; then
echo "❌ Configuration file does not exist: $PROXMOX_MCP_CONFIG"
echo "Please ensure the configuration file is properly set up"
exit 1
fi
echo "✅ Configuration file: $PROXMOX_MCP_CONFIG"
echo "✅ Virtual environment activated"
echo ""
echo "🔍 Starting server..."
echo "Press Ctrl+C to stop the server"
echo ""
# Start server
python -m proxmox_mcp.server
```
--------------------------------------------------------------------------------
/test_scripts/test_vm_start.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test VM startup functionality
"""
import os
import sys
def test_start_vm_101():
"""Test starting VM 101 (VPN-Server)"""
# Set configuration
os.environ['PROXMOX_MCP_CONFIG'] = 'proxmox-config/config.json'
try:
from proxmox_mcp.config.loader import load_config
from proxmox_mcp.core.proxmox import ProxmoxManager
from proxmox_mcp.tools.vm import VMTools
config = load_config('proxmox-config/config.json')
manager = ProxmoxManager(config.proxmox, config.auth)
api = manager.get_api()
vm_tools = VMTools(api)
print("🚀 Test starting VPN-Server (VM 101)")
print("=" * 50)
# Start VM 101
result = vm_tools.start_vm(node="pve", vmid="101")
for content in result:
print(content.text)
return True
except Exception as e:
print(f"❌ Start failed: {e}")
return False
if __name__ == "__main__":
print("🔍 Test VM startup functionality")
print("=" * 50)
success = test_start_vm_101()
if success:
print("\n✅ Test completed")
else:
print("\n❌ Test failed")
sys.exit(1)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/utils/logging.py:
--------------------------------------------------------------------------------
```python
"""
Logging configuration for the Proxmox MCP server.
"""
import logging
import sys
from typing import Optional
def setup_logging(
level: str = "INFO",
format_str: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
log_file: Optional[str] = None,
) -> logging.Logger:
"""
Configure logging for the Proxmox MCP server.
Args:
level: The logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
format_str: The format string for log messages
log_file: Optional file path to write logs to
Returns:
logging.Logger: Configured logger instance
"""
# Create logger
logger = logging.getLogger("proxmox-mcp")
logger.setLevel(getattr(logging, level.upper()))
# Create handlers
handlers = []
# Console handler
console_handler = logging.StreamHandler(sys.stderr)
console_handler.setLevel(getattr(logging, level.upper()))
handlers.append(console_handler)
# File handler if log_file is specified
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(getattr(logging, level.upper()))
handlers.append(file_handler)
# Create formatter
formatter = logging.Formatter(format_str)
# Add formatter to handlers and handlers to logger
for handler in handlers:
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
```
--------------------------------------------------------------------------------
/start_openapi.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Proxmox MCP OpenAPI startup script
# Expose MCP server as OpenAPI REST endpoints through mcpo proxy
# Configurable deployment address
# Get host and port from environment variables or use defaults
HOST=${OPENAPI_HOST:-"localhost"}
PORT=${OPENAPI_PORT:-"8811"}
echo "🛰️ Starting Proxmox MCP OpenAPI server..."
echo ""
# Check if mcpo is installed
if ! command -v mcpo &> /dev/null; then
echo "❌ mcpo not installed, installing..."
pip install mcpo
fi
# Check virtual environment
if [ ! -d ".venv" ]; then
echo "❌ Virtual environment does not exist, please run installation steps first"
exit 1
fi
# Check configuration file
if [ ! -f "proxmox-config/config.json" ]; then
echo "❌ Configuration file does not exist: proxmox-config/config.json"
echo "Please ensure the configuration file is properly set up"
exit 1
fi
echo "✅ Configuration file: proxmox-config/config.json"
echo "✅ mcpo proxy ready"
echo ""
echo "🚀 Starting OpenAPI proxy server..."
echo "🌐 Service address: http://${HOST}:${PORT}"
echo "📖 API documentation: http://${HOST}:${PORT}/docs"
echo "🔧 OpenAPI specification: http://${HOST}:${PORT}/openapi.json"
echo "❤️ Health check: http://${HOST}:${PORT}/health"
echo ""
echo "Press Ctrl+C to stop the server"
echo ""
echo "💡 To use a different host/port, set environment variables:"
echo " export OPENAPI_HOST=your-host"
echo " export OPENAPI_PORT=your-port"
echo ""
# Set environment variables
export PROXMOX_MCP_CONFIG="$(pwd)/proxmox-config/config.json"
# Start mcpo proxy server, bind to all interfaces on specified port
mcpo --host 0.0.0.0 --port ${PORT} -- bash -c "cd $(pwd) && source .venv/bin/activate && python -m proxmox_mcp.server"
```
--------------------------------------------------------------------------------
/test_scripts/test_common.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Common configuration helper for test scripts
"""
import os
import sys
from pathlib import Path
def setup_test_environment():
"""Set up test environment configuration paths"""
# Get current script directory
current_dir = Path(__file__).parent
# Calculate project root directory
project_root = current_dir.parent
# Set configuration file path
config_path = project_root / "proxmox-config" / "config.json"
# Set source code path
src_path = project_root / "src"
# Ensure paths exist
if not config_path.exists():
raise FileNotFoundError(f"Configuration file does not exist: {config_path}")
if not src_path.exists():
raise FileNotFoundError(f"Source code directory does not exist: {src_path}")
# Set environment variables
os.environ['PROXMOX_MCP_CONFIG'] = str(config_path)
# Add source code path to Python path
if str(src_path) not in sys.path:
sys.path.insert(0, str(src_path))
return str(config_path)
def get_test_tools():
"""Get test tools classes"""
# Ensure environment is set up
config_path = setup_test_environment()
try:
from proxmox_mcp.config.loader import load_config
from proxmox_mcp.core.proxmox import ProxmoxManager
from proxmox_mcp.tools.vm import VMTools
config = load_config(config_path)
manager = ProxmoxManager(config.proxmox, config.auth)
api = manager.get_api()
vm_tools = VMTools(api)
return {
'config': config,
'manager': manager,
'api': api,
'vm_tools': vm_tools
}
except Exception as e:
print(f"❌ Failed to initialize test tools: {e}")
raise
def print_test_header(title):
"""Print test title"""
print(f"🔍 {title}")
print("=" * len(f"🔍 {title}"))
def print_test_result(success, message=""):
"""Print test result"""
if success:
print(f"\n✅ Test completed {message}")
else:
print(f"\n❌ Test failed {message}")
sys.exit(1)
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=61.0.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "proxmox-mcp"
version = "0.1.0"
description = "A Model Context Protocol server for interacting with Proxmox hypervisors"
requires-python = ">=3.11"
authors = [
{name = "Kevin", email = "[email protected]"}
]
readme = "README.md"
license = {text = "MIT"}
keywords = ["proxmox", "mcp", "virtualization", "cline", "qemu", "lxc"]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: System :: Systems Administration",
]
dependencies = [
"mcp @ git+https://github.com/modelcontextprotocol/python-sdk.git",
"proxmoxer>=2.0.1,<3.0.0",
"requests>=2.31.0,<3.0.0",
"pydantic>=2.0.0,<3.0.0",
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
"mcpo>=0.0.17",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0,<8.0.0",
"black>=23.0.0,<24.0.0",
"mypy>=1.0.0,<2.0.0",
"pytest-asyncio>=0.21.0,<0.22.0",
"ruff>=0.1.0,<0.2.0",
"types-requests>=2.31.0,<3.0.0",
]
[project.urls]
Homepage = "https://github.com/yourusername/proxmox-mcp"
Documentation = "https://github.com/yourusername/proxmox-mcp#readme"
Repository = "https://github.com/yourusername/proxmox-mcp.git"
Issues = "https://github.com/yourusername/proxmox-mcp/issues"
[project.scripts]
proxmox-mcp = "proxmox_mcp.server:main"
[tool.pytest.ini_options]
asyncio_mode = "strict"
testpaths = ["tests"]
python_files = ["test_*.py"]
addopts = "-v"
[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
[tool.ruff]
select = ["E", "F", "B", "I"]
ignore = []
line-length = 100
target-version = "py39"
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/utils/auth.py:
--------------------------------------------------------------------------------
```python
"""
Authentication utilities for the Proxmox MCP server.
"""
import os
from typing import Dict, Optional, Tuple
from pydantic import BaseModel
class ProxmoxAuth(BaseModel):
"""Proxmox authentication configuration."""
user: str
token_name: str
token_value: str
def load_auth_from_env() -> ProxmoxAuth:
"""
Load Proxmox authentication details from environment variables.
Environment Variables:
PROXMOX_USER: Username with realm (e.g., 'root@pam' or 'user@pve')
PROXMOX_TOKEN_NAME: API token name
PROXMOX_TOKEN_VALUE: API token value
Returns:
ProxmoxAuth: Authentication configuration
Raises:
ValueError: If required environment variables are missing
"""
user = os.getenv("PROXMOX_USER")
token_name = os.getenv("PROXMOX_TOKEN_NAME")
token_value = os.getenv("PROXMOX_TOKEN_VALUE")
if not all([user, token_name, token_value]):
missing = []
if not user:
missing.append("PROXMOX_USER")
if not token_name:
missing.append("PROXMOX_TOKEN_NAME")
if not token_value:
missing.append("PROXMOX_TOKEN_VALUE")
raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
return ProxmoxAuth(
user=user,
token_name=token_name,
token_value=token_value,
)
def parse_user(user: str) -> Tuple[str, str]:
"""
Parse a Proxmox user string into username and realm.
Args:
user: User string in format 'username@realm'
Returns:
Tuple[str, str]: (username, realm)
Raises:
ValueError: If user string is not in correct format
"""
try:
username, realm = user.split("@")
return username, realm
except ValueError:
raise ValueError(
"Invalid user format. Expected 'username@realm' (e.g., 'root@pam' or 'user@pve')"
)
def get_auth_dict(auth: ProxmoxAuth) -> Dict[str, str]:
"""
Convert ProxmoxAuth model to dictionary for Proxmoxer API.
Args:
auth: ProxmoxAuth configuration
Returns:
Dict[str, str]: Authentication dictionary for Proxmoxer
"""
return {
"user": auth.user,
"token_name": auth.token_name,
"token_value": auth.token_value,
}
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/config/loader.py:
--------------------------------------------------------------------------------
```python
"""
Configuration loading utilities for the Proxmox MCP server.
This module handles loading and validation of server configuration:
- JSON configuration file loading
- Environment variable handling
- Configuration validation using Pydantic models
- Error handling for invalid configurations
The module ensures that all required configuration is present
and valid before the server starts operation.
"""
import json
import os
from typing import Optional
from .models import Config
def load_config(config_path: Optional[str] = None) -> Config:
"""Load and validate configuration from JSON file.
Performs the following steps:
1. Verifies config path is provided
2. Loads JSON configuration file
3. Validates required fields are present
4. Converts to typed Config object using Pydantic
Configuration must include:
- Proxmox connection settings (host, port, etc.)
- Authentication credentials (user, token)
- Logging configuration
Args:
config_path: Path to the JSON configuration file
If not provided, raises ValueError
Returns:
Config object containing validated configuration:
{
"proxmox": {
"host": "proxmox-host",
"port": 8006,
...
},
"auth": {
"user": "username",
"token_name": "token-name",
...
},
"logging": {
"level": "INFO",
...
}
}
Raises:
ValueError: If:
- Config path is not provided
- JSON is invalid
- Required fields are missing
- Field values are invalid
"""
if not config_path:
raise ValueError("PROXMOX_MCP_CONFIG environment variable must be set")
try:
with open(config_path) as f:
config_data = json.load(f)
if not config_data.get('proxmox', {}).get('host'):
raise ValueError("Proxmox host cannot be empty")
return Config(**config_data)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in config file: {e}")
except Exception as e:
raise ValueError(f"Failed to load config: {e}")
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
"""
Setup script for the Proxmox MCP server.
This file is maintained for compatibility with older tools.
For modern Python packaging, see pyproject.toml.
"""
from setuptools import setup, find_packages
# Metadata and dependencies are primarily managed in pyproject.toml
# This file exists for compatibility with tools that don't support pyproject.toml
setup(
name="proxmox-mcp",
version="0.1.0",
packages=find_packages(where="src"),
package_dir={"": "src"},
python_requires=">=3.9",
install_requires=[
"mcp @ git+https://github.com/modelcontextprotocol/python-sdk.git",
"proxmoxer>=2.0.1,<3.0.0",
"requests>=2.31.0,<3.0.0",
"pydantic>=2.0.0,<3.0.0",
],
extras_require={
"dev": [
"pytest>=7.0.0,<8.0.0",
"black>=23.0.0,<24.0.0",
"mypy>=1.0.0,<2.0.0",
"pytest-asyncio>=0.21.0,<0.22.0",
"ruff>=0.1.0,<0.2.0",
"types-requests>=2.31.0,<3.0.0",
],
},
entry_points={
"console_scripts": [
"proxmox-mcp=proxmox_mcp.server:main",
],
},
author="Kevin",
author_email="[email protected]",
description="A Model Context Protocol server for interacting with Proxmox hypervisors",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
license="MIT",
keywords=["proxmox", "mcp", "virtualization", "cline", "qemu", "lxc"],
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: System :: Systems Administration",
"Topic :: System :: Virtualization",
],
project_urls={
"Homepage": "https://github.com/yourusername/proxmox-mcp",
"Documentation": "https://github.com/yourusername/proxmox-mcp#readme",
"Repository": "https://github.com/yourusername/proxmox-mcp.git",
"Issues": "https://github.com/yourusername/proxmox-mcp/issues",
},
)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/theme.py:
--------------------------------------------------------------------------------
```python
"""
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'])
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/core/logging.py:
--------------------------------------------------------------------------------
```python
"""
Logging configuration for the Proxmox MCP server.
This module handles logging setup and configuration:
- File and console logging handlers
- Log level management
- Format customization
- Handler lifecycle management
The logging system supports:
- Configurable log levels
- File-based logging with path resolution
- Console logging for errors
- Custom format strings
- Multiple handler management
"""
import logging
import os
from typing import Optional
from ..config.models import LoggingConfig
def setup_logging(config: LoggingConfig) -> logging.Logger:
"""Configure and initialize logging system.
Sets up a comprehensive logging system with:
- File logging (if configured):
* Handles relative/absolute paths
* Uses configured log level
* Applies custom format
- Console logging:
* Always enabled for errors
* Ensures critical issues are visible
- Handler Management:
* Removes existing handlers
* Configures new handlers
* Sets up formatters
Args:
config: Logging configuration containing:
- Log level (e.g., "INFO", "DEBUG")
- Format string
- Optional log file path
Returns:
Configured logger instance for "proxmox-mcp"
with appropriate handlers and formatting
Example config:
{
"level": "INFO",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"file": "/path/to/log/file.log" # Optional
}
"""
# Convert relative path to absolute
log_file = config.file
if log_file and not os.path.isabs(log_file):
log_file = os.path.join(os.getcwd(), log_file)
# Create handlers
handlers = []
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(getattr(logging, config.level.upper()))
handlers.append(file_handler)
# Console handler for errors only
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.ERROR)
handlers.append(console_handler)
# Configure formatters
formatter = logging.Formatter(config.format)
for handler in handlers:
handler.setFormatter(formatter)
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, config.level.upper()))
# Remove any existing handlers
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# Add new handlers
for handler in handlers:
root_logger.addHandler(handler)
# Create and return server logger
logger = logging.getLogger("proxmox-mcp")
return logger
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/cluster.py:
--------------------------------------------------------------------------------
```python
"""
Cluster-related tools for Proxmox MCP.
This module provides tools for monitoring and managing Proxmox clusters:
- Retrieving overall cluster health status
- Monitoring quorum status and node count
- Tracking cluster resources and configuration
- Checking cluster-wide service availability
The tools provide essential information for maintaining
cluster health and ensuring proper operation.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_CLUSTER_STATUS_DESC
class ClusterTools(ProxmoxTool):
"""Tools for managing Proxmox cluster.
Provides functionality for:
- Monitoring cluster health and status
- Tracking quorum and node membership
- Managing cluster-wide resources
- Verifying cluster configuration
Essential for maintaining cluster health and ensuring
proper operation of the Proxmox environment.
"""
def get_cluster_status(self) -> List[Content]:
"""Get overall Proxmox cluster health and configuration status.
Retrieves comprehensive cluster information including:
- Cluster name and identity
- Quorum status (essential for cluster operations)
- Active node count and health
- Resource distribution and status
This information is critical for:
- Ensuring cluster stability
- Monitoring node membership
- Verifying resource availability
- Detecting potential issues
Returns:
List of Content objects containing formatted cluster status:
{
"name": "cluster-name",
"quorum": true/false,
"nodes": count,
"resources": [
{
"type": "resource-type",
"status": "status"
}
]
}
Raises:
RuntimeError: If cluster status query fails due to:
- Network connectivity issues
- Authentication problems
- API endpoint failures
"""
try:
result = self.proxmox.cluster.status.get()
first_item = result[0] if result and len(result) > 0 else {}
status = {
"name": first_item.get("name") if first_item else None,
"quorum": first_item.get("quorate") if first_item else None,
"nodes": len([node for node in result if node.get("type") == "node"]) if result else 0,
"resources": [res for res in result if res.get("type") == "resource"] if result else []
}
return self._format_response(status, "cluster")
except Exception as e:
self._handle_error("get cluster status", e)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/config/models.py:
--------------------------------------------------------------------------------
```python
"""
Configuration models for the Proxmox MCP server.
This module defines Pydantic models for configuration validation:
- Proxmox connection settings
- Authentication credentials
- Logging configuration
- Tool-specific parameter models
The models provide:
- Type validation
- Default values
- Field descriptions
- Required vs optional field handling
"""
from typing import Optional, Annotated
from pydantic import BaseModel, Field
class NodeStatus(BaseModel):
"""Model for node status query parameters.
Validates and documents the required parameters for
querying a specific node's status in the cluster.
"""
node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1', 'proxmox-node2')")]
class VMCommand(BaseModel):
"""Model for VM command execution parameters.
Validates and documents the required parameters for
executing commands within a VM via QEMU guest agent.
"""
node: Annotated[str, Field(description="Host node name (e.g. 'pve1', 'proxmox-node2')")]
vmid: Annotated[str, Field(description="VM ID number (e.g. '100', '101')")]
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a', 'systemctl status nginx')")]
class ProxmoxConfig(BaseModel):
"""Model for Proxmox connection configuration.
Defines the required and optional parameters for
establishing a connection to the Proxmox API server.
Provides sensible defaults for optional parameters.
"""
host: str # Required: Proxmox host address
port: int = 8006 # Optional: API port (default: 8006)
verify_ssl: bool = True # Optional: SSL verification (default: True)
service: str = "PVE" # Optional: Service type (default: PVE)
class AuthConfig(BaseModel):
"""Model for Proxmox authentication configuration.
Defines the required parameters for API authentication
using token-based authentication. All fields are required
to ensure secure API access.
"""
user: str # Required: Username (e.g., 'root@pam')
token_name: str # Required: API token name
token_value: str # Required: API token secret
class LoggingConfig(BaseModel):
"""Model for logging configuration.
Defines logging parameters with sensible defaults.
Supports both file and console logging with
customizable format and log levels.
"""
level: str = "INFO" # Optional: Log level (default: INFO)
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" # Optional: Log format
file: Optional[str] = None # Optional: Log file path (default: None for console logging)
class Config(BaseModel):
"""Root configuration model.
Combines all configuration models into a single validated
configuration object. All sections are required to ensure
proper server operation.
"""
proxmox: ProxmoxConfig # Required: Proxmox connection settings
auth: AuthConfig # Required: Authentication credentials
logging: LoggingConfig # Required: Logging configuration
```
--------------------------------------------------------------------------------
/tests/test_vm_console.py:
--------------------------------------------------------------------------------
```python
"""
Tests for VM console operations.
"""
import pytest
from unittest.mock import Mock, patch
from proxmox_mcp.tools.console import VMConsoleManager
@pytest.fixture
def mock_proxmox():
"""Fixture to create a mock ProxmoxAPI instance."""
mock = Mock()
# Setup chained mock calls
mock.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "running"
}
mock.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
"out": "command output",
"err": "",
"exitcode": 0
}
return mock
@pytest.fixture
def vm_console(mock_proxmox):
"""Fixture to create a VMConsoleManager instance."""
return VMConsoleManager(mock_proxmox)
@pytest.mark.asyncio
async def test_execute_command_success(vm_console, mock_proxmox):
"""Test successful command execution."""
result = await vm_console.execute_command("node1", "100", "ls -l")
assert result["success"] is True
assert result["output"] == "command output"
assert result["error"] == ""
assert result["exit_code"] == 0
# Verify correct API calls
mock_proxmox.nodes.return_value.qemu.assert_called_with("100")
mock_proxmox.nodes.return_value.qemu.return_value.agent.exec.post.assert_called_with(
command="ls -l"
)
@pytest.mark.asyncio
async def test_execute_command_vm_not_running(vm_console, mock_proxmox):
"""Test command execution on stopped VM."""
mock_proxmox.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "stopped"
}
with pytest.raises(ValueError, match="not running"):
await vm_console.execute_command("node1", "100", "ls -l")
@pytest.mark.asyncio
async def test_execute_command_vm_not_found(vm_console, mock_proxmox):
"""Test command execution on non-existent VM."""
mock_proxmox.nodes.return_value.qemu.return_value.status.current.get.side_effect = \
Exception("VM not found")
with pytest.raises(ValueError, match="not found"):
await vm_console.execute_command("node1", "100", "ls -l")
@pytest.mark.asyncio
async def test_execute_command_failure(vm_console, mock_proxmox):
"""Test command execution failure."""
mock_proxmox.nodes.return_value.qemu.return_value.agent.exec.post.side_effect = \
Exception("Command failed")
with pytest.raises(RuntimeError, match="Failed to execute command"):
await vm_console.execute_command("node1", "100", "ls -l")
@pytest.mark.asyncio
async def test_execute_command_with_error_output(vm_console, mock_proxmox):
"""Test command execution with error output."""
mock_proxmox.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
"out": "",
"err": "command error",
"exitcode": 1
}
result = await vm_console.execute_command("node1", "100", "invalid-command")
assert result["success"] is True # Success refers to API call, not command
assert result["output"] == ""
assert result["error"] == "command error"
assert result["exit_code"] == 1
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/colors.py:
--------------------------------------------------------------------------------
```python
"""
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
```
--------------------------------------------------------------------------------
/test_scripts/test_create_vm.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test VM creation functionality
"""
import os
import sys
def test_create_vm():
"""Test creating VM - 1 CPU, 2GB RAM, 10GB storage"""
# Set configuration
os.environ['PROXMOX_MCP_CONFIG'] = 'proxmox-config/config.json'
try:
from proxmox_mcp.config.loader import load_config
from proxmox_mcp.core.proxmox import ProxmoxManager
from proxmox_mcp.tools.vm import VMTools
config = load_config('proxmox-config/config.json')
manager = ProxmoxManager(config.proxmox, config.auth)
api = manager.get_api()
vm_tools = VMTools(api)
print("🎉 Test creating new VM - user requested configuration")
print("=" * 60)
print("Configuration:")
print(" • CPU: 1 core")
print(" • RAM: 2 GB (2048 MB)")
print(" • Storage: 10 GB")
print(" • VM ID: 999 (test purpose)")
print(" • Name: test-vm-demo")
print()
# Find an available VM ID
vmid = "999"
# Check if VM ID already exists
try:
existing_vm = api.nodes("pve").qemu(vmid).config.get()
print(f"⚠️ VM {vmid} already exists, will try VM ID 998")
vmid = "998"
existing_vm = api.nodes("pve").qemu(vmid).config.get()
print(f"⚠️ VM {vmid} also exists, will try VM ID 997")
vmid = "997"
except:
print(f"✅ VM ID {vmid} is available")
# Create VM
result = vm_tools.create_vm(
node="pve",
vmid=vmid,
name="test-vm-demo",
cpus=1,
memory=2048, # 2GB in MB
disk_size=10 # 10GB
)
for content in result:
print(content.text)
return True
except Exception as e:
print(f"❌ Creation failed: {e}")
return False
def test_list_vms():
"""Test listing VMs to confirm successful creation"""
os.environ['PROXMOX_MCP_CONFIG'] = 'proxmox-config/config.json'
try:
from proxmox_mcp.config.loader import load_config
from proxmox_mcp.core.proxmox import ProxmoxManager
from proxmox_mcp.tools.vm import VMTools
config = load_config('proxmox-config/config.json')
manager = ProxmoxManager(config.proxmox, config.auth)
api = manager.get_api()
vm_tools = VMTools(api)
print("\n🔍 List all VMs to confirm creation results:")
print("=" * 40)
result = vm_tools.get_vms()
for content in result:
# Only show newly created VM information
lines = content.text.split('\n')
for line in lines:
if 'test-vm-demo' in line or 'VM 99' in line:
print(line)
return True
except Exception as e:
print(f"❌ List query failed: {e}")
return False
if __name__ == "__main__":
print("🔍 Test VM creation functionality")
print("=" * 50)
success = test_create_vm()
if success:
print("\n✅ Creation test completed")
# Test listing VMs
test_list_vms()
else:
print("\n❌ Creation test failed")
sys.exit(1)
```
--------------------------------------------------------------------------------
/test_scripts/test_vm_power.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test Proxmox VM power management functionality
"""
import sys
from test_common import setup_test_environment, get_test_tools, print_test_header, print_test_result
def test_vm_power_operations():
"""Test VM power management operations"""
try:
# Set up test environment
setup_test_environment()
tools = get_test_tools()
api = tools['api']
nodes = api.nodes.get()
# Safely get the first node to avoid index out of range error
if not nodes or len(nodes) == 0:
print("❌ No Proxmox nodes found")
return False
node_name = nodes[0]['node']
print(f"Test node: {node_name}")
# Get all VMs
vms = api.nodes(node_name).qemu.get()
print(f"Found {len(vms)} virtual machines:")
vm_101_found = False
for vm in vms:
vmid = vm['vmid']
name = vm['name']
status = vm['status']
print(f" - VM {vmid}: {name} ({status})")
if vmid == 101:
vm_101_found = True
print(f"\nFound VPN-Server (ID: 101), current status: {status}")
# Test available status operations
vm_api = api.nodes(node_name).qemu(vmid)
status_api = vm_api.status
print("Test available status operations:")
# Try accessing different status endpoints
try:
# Check if start endpoint exists
if hasattr(status_api, 'start'):
print(" ✅ Supports start operation")
else:
print(" ❌ Does not support start operation")
if hasattr(status_api, 'stop'):
print(" ✅ Supports stop operation")
else:
print(" ❌ Does not support stop operation")
if hasattr(status_api, 'reset'):
print(" ✅ Supports reset operation")
else:
print(" ❌ Does not support reset operation")
if hasattr(status_api, 'shutdown'):
print(" ✅ Supports shutdown operation")
else:
print(" ❌ Does not support shutdown operation")
# If VM is stopped, try to start
if status == 'stopped':
print(f"\nVM {vmid} is currently stopped, can try to start")
print("Start command would be: api.nodes(node).qemu(101).status.start.post()")
elif status == 'running':
print(f"\nVM {vmid} is currently running")
except Exception as e:
print(f" Error while testing operations: {e}")
if not vm_101_found:
print("\n❌ VM 101 (VPN-Server) not found")
except Exception as e:
print(f"Test failed: {e}")
return False
return True
if __name__ == "__main__":
print_test_header("Test Proxmox VM power management functionality")
success = test_vm_power_operations()
print_test_result(success)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/storage.py:
--------------------------------------------------------------------------------
```python
"""
Storage-related tools for Proxmox MCP.
This module provides tools for managing and monitoring Proxmox storage:
- Listing all storage pools across the cluster
- Retrieving detailed storage information including:
* Storage type and content types
* Usage statistics and capacity
* Availability status
* Node assignments
The tools implement fallback mechanisms for scenarios where
detailed storage information might be temporarily unavailable.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_STORAGE_DESC
class StorageTools(ProxmoxTool):
"""Tools for managing Proxmox storage.
Provides functionality for:
- Retrieving cluster-wide storage information
- Monitoring storage pool status and health
- Tracking storage utilization and capacity
- Managing storage content types
Implements fallback mechanisms for scenarios where detailed
storage information might be temporarily unavailable.
"""
def get_storage(self) -> List[Content]:
"""List storage pools across the cluster with detailed status.
Retrieves comprehensive information for each storage pool including:
- Basic identification (name, type)
- Content types supported (VM disks, backups, ISO images, etc.)
- Availability status (online/offline)
- Usage statistics:
* Used space
* Total capacity
* Available space
Implements a fallback mechanism that returns basic information
if detailed status retrieval fails for any storage pool.
Returns:
List of Content objects containing formatted storage information:
{
"storage": "storage-name",
"type": "storage-type",
"content": ["content-types"],
"status": "online/offline",
"used": bytes,
"total": bytes,
"available": bytes
}
Raises:
RuntimeError: If the cluster-wide storage query fails
"""
try:
result = self.proxmox.storage.get()
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)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/core/proxmox.py:
--------------------------------------------------------------------------------
```python
"""
Proxmox API setup and management.
This module handles the core Proxmox API integration, providing:
- Secure API connection setup and management
- Token-based authentication
- Connection testing and validation
- Error handling for API operations
The ProxmoxManager class serves as the central point for all Proxmox API
interactions, ensuring consistent connection handling and authentication
across the MCP server.
"""
import logging
from typing import Dict, Any
from proxmoxer import ProxmoxAPI
from ..config.models import ProxmoxConfig, AuthConfig
class ProxmoxManager:
"""Manager class for Proxmox API operations.
This class handles:
- API connection initialization and management
- Configuration validation and merging
- Connection testing and health checks
- Token-based authentication setup
The manager provides a single point of access to the Proxmox API,
ensuring proper initialization and error handling for all API operations.
"""
def __init__(self, proxmox_config: ProxmoxConfig, auth_config: AuthConfig):
"""Initialize the Proxmox API manager.
Args:
proxmox_config: Proxmox connection configuration
auth_config: Authentication configuration
"""
self.logger = logging.getLogger("proxmox-mcp.proxmox")
self.config = self._create_config(proxmox_config, auth_config)
self.api = self._setup_api()
def _create_config(self, proxmox_config: ProxmoxConfig, auth_config: AuthConfig) -> Dict[str, Any]:
"""Create a configuration dictionary for ProxmoxAPI.
Merges connection and authentication configurations into a single
dictionary suitable for ProxmoxAPI initialization. Handles:
- Host and port configuration
- SSL verification settings
- Token-based authentication details
- Service type specification
Args:
proxmox_config: Proxmox connection configuration (host, port, SSL settings)
auth_config: Authentication configuration (user, token details)
Returns:
Dictionary containing merged configuration ready for API initialization
"""
return {
'host': proxmox_config.host,
'port': proxmox_config.port,
'user': auth_config.user,
'token_name': auth_config.token_name,
'token_value': auth_config.token_value,
'verify_ssl': proxmox_config.verify_ssl,
'service': proxmox_config.service
}
def _setup_api(self) -> ProxmoxAPI:
"""Initialize and test Proxmox API connection.
Performs the following steps:
1. Creates ProxmoxAPI instance with configured settings
2. Tests connection by making a version check request
3. Validates authentication and permissions
4. Logs connection status and any issues
Returns:
Initialized and tested ProxmoxAPI instance
Raises:
RuntimeError: If connection fails due to:
- Invalid host/port
- Authentication failure
- Network connectivity issues
- SSL certificate validation errors
"""
try:
self.logger.info(f"Connecting to Proxmox host: {self.config['host']}")
api = ProxmoxAPI(**self.config)
# Test connection
api.version.get()
self.logger.info("Successfully connected to Proxmox API")
return api
except Exception as e:
self.logger.error(f"Failed to connect to Proxmox: {e}")
raise RuntimeError(f"Failed to connect to Proxmox: {e}")
def get_api(self) -> ProxmoxAPI:
"""Get the initialized Proxmox API instance.
Provides access to the configured and tested ProxmoxAPI instance
for making API calls. The instance maintains connection state and
handles authentication automatically.
Returns:
ProxmoxAPI instance ready for making API calls
"""
return self.api
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/base.py:
--------------------------------------------------------------------------------
```python
"""
Base classes and utilities for Proxmox MCP tools.
This module provides the foundation for all Proxmox MCP tools, including:
- Base tool class with common functionality
- Response formatting utilities
- Error handling mechanisms
- Logging setup
All tool implementations inherit from the ProxmoxTool base class to ensure
consistent behavior and error handling across the MCP server.
"""
import logging
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.
This class provides common functionality used by all Proxmox tool implementations:
- Proxmox API access
- Standardized logging
- Response formatting
- Error handling
All tool classes should inherit from this base class to ensure consistent
behavior and error handling across the MCP server.
"""
def __init__(self, proxmox_api: ProxmoxAPI):
"""Initialize the tool.
Args:
proxmox_api: Initialized ProxmoxAPI instance
"""
self.proxmox = proxmox_api
self.logger = logging.getLogger(f"proxmox-mcp.{self.__class__.__name__.lower()}")
def _format_response(self, data: Any, resource_type: Optional[str] = None) -> List[Content]:
"""Format response data into MCP content using templates.
This method handles formatting of various Proxmox resource types into
consistent MCP content responses. It uses specialized templates for
different resource types (nodes, VMs, storage, etc.) and falls back
to JSON formatting for unknown types.
Args:
data: Raw data from Proxmox API to format
resource_type: Type of resource for template selection. Valid types:
'nodes', 'node_status', 'vms', 'storage', 'containers', 'cluster'
Returns:
List of Content objects formatted according to resource type
"""
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)
elif resource_type == "cluster":
formatted = ProxmoxTemplates.cluster_status(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 from Proxmox operations.
Provides standardized error handling across all tools by:
- Logging errors with appropriate context
- Categorizing errors into specific exception types
- Converting Proxmox-specific errors into standard Python exceptions
Args:
operation: Description of the operation that failed (e.g., "get node status")
error: The exception that occurred during the operation
Raises:
ValueError: For invalid input, missing resources, or permission issues
RuntimeError: For unexpected errors or API failures
"""
error_msg = str(error)
self.logger.error(f"Failed to {operation}: {error_msg}")
if "not found" in error_msg.lower():
raise ValueError(f"Resource not found: {error_msg}")
if "permission denied" in error_msg.lower():
raise ValueError(f"Permission denied: {error_msg}")
if "invalid" in error_msg.lower():
raise ValueError(f"Invalid input: {error_msg}")
raise RuntimeError(f"Failed to {operation}: {error_msg}")
```
--------------------------------------------------------------------------------
/test_scripts/test_openapi.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test OpenAPI functionality
"""
import requests
import json
import os
# Get base URL from environment variable or use default localhost
BASE_URL = os.getenv('OPENAPI_BASE_URL', 'http://localhost:8811')
def test_basic_endpoints():
"""Test basic API endpoints"""
print("🔍 Test basic API endpoints")
print(f"🌐 Using base URL: {BASE_URL}")
print("=" * 50)
# Test get nodes
try:
response = requests.post(f"{BASE_URL}/get_nodes")
print(f"✅ get_nodes: {response.status_code} - {len(response.text)} chars")
except Exception as e:
print(f"❌ get_nodes error: {e}")
# Test get VM list
try:
response = requests.post(f"{BASE_URL}/get_vms")
print(f"✅ get_vms: {response.status_code} - {len(response.text)} chars")
if response.status_code == 200:
# Check if our test VMs are included
if "test-vm" in response.text:
print(" 📋 Test VM found")
except Exception as e:
print(f"❌ get_vms error: {e}")
def test_vm_creation_api():
"""Test VM creation API"""
print("\n🎉 Test VM creation API - user requested configuration")
print("=" * 50)
print("Configuration: 1 CPU core, 2GB RAM, 10GB storage")
# VM creation parameters
create_data = {
"node": "pve",
"vmid": "996", # Use new VM ID
"name": "user-requested-vm",
"cpus": 1,
"memory": 2048, # 2GB in MB
"disk_size": 10 # 10GB
}
try:
response = requests.post(
f"{BASE_URL}/create_vm",
json=create_data,
headers={"Content-Type": "application/json"}
)
print(f"📡 API response status: {response.status_code}")
if response.status_code == 200:
result = response.json()
print("✅ VM creation successful!")
print(f"📄 Response content: {json.dumps(result, indent=2, ensure_ascii=False)}")
else:
print(f"❌ VM creation failed: {response.text}")
except requests.exceptions.ConnectionError:
print("❌ Cannot connect to API server - please ensure OpenAPI service is running")
except Exception as e:
print(f"❌ API call error: {e}")
def test_vm_power_api():
"""Test VM power management API"""
print("\n🚀 Test VM power management API")
print("=" * 50)
# Test starting VM 101 (VPN-Server)
start_data = {
"node": "pve",
"vmid": "101"
}
try:
response = requests.post(
f"{BASE_URL}/start_vm",
json=start_data,
headers={"Content-Type": "application/json"}
)
print(f"📡 Start VM 101 response: {response.status_code}")
if response.status_code == 200:
result = response.json()
print("✅ VM start command successful!")
print(f"📄 Response: {json.dumps(result, indent=2, ensure_ascii=False)}")
else:
print(f"❌ VM start failed: {response.text}")
except requests.exceptions.ConnectionError:
print("❌ Cannot connect to API server")
except Exception as e:
print(f"❌ API call error: {e}")
def list_available_apis():
"""List all available API endpoints"""
print("\n📋 Available API endpoints")
print("=" * 50)
try:
response = requests.get(f"{BASE_URL}/openapi.json")
if response.status_code == 200:
openapi_spec = response.json()
paths = openapi_spec.get("paths", {})
print(f"Found {len(paths)} API endpoints:")
for path, methods in paths.items():
for method, details in methods.items():
summary = details.get("summary", "No summary")
print(f" • {method.upper()} {path} - {summary}")
else:
print(f"❌ Cannot get API specification: {response.status_code}")
except Exception as e:
print(f"❌ Get API list error: {e}")
if __name__ == "__main__":
print("🔍 ProxmoxMCP OpenAPI functionality test")
print("=" * 60)
# List available APIs
list_available_apis()
# Test basic functionality
test_basic_endpoints()
# Test VM creation functionality
test_vm_creation_api()
# Test VM power management
test_vm_power_api()
print("\n✅ All tests completed")
print("\n💡 Usage instructions:")
print("When user says 'Can you create a VM with 1 cpu core and 2 GB ram with 10GB of storage disk',")
print("the AI assistant can call create_vm API to complete the task!")
print(f"\n🔧 To test with different server, set environment variable:")
print("export OPENAPI_BASE_URL=http://your-server:8811")
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/formatters.py:
--------------------------------------------------------------------------------
```python
"""
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
"""
result = [
f"{ProxmoxTheme.ACTIONS['command']} Console Command Result",
f" • Status: {'SUCCESS' if success else 'FAILED'}",
f" • Command: {command}",
"",
"Output:",
output.strip()
]
if error:
result.extend([
"",
"Error:",
error.strip()
])
return "\n".join(result)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/definitions.py:
--------------------------------------------------------------------------------
```python
"""
Tool descriptions for Proxmox MCP tools.
"""
# Node tool descriptions
GET_NODES_DESC = """List all nodes in the Proxmox cluster with their status, CPU, memory, and role information.
Example:
{"node": "pve1", "status": "online", "cpu_usage": 0.15, "memory": {"used": "8GB", "total": "32GB"}}"""
GET_NODE_STATUS_DESC = """Get detailed status information for a specific Proxmox node.
Parameters:
node* - Name/ID of node to query (e.g. 'pve1')
Example:
{"cpu": {"usage": 0.15}, "memory": {"used": "8GB", "total": "32GB"}}"""
# VM tool descriptions
GET_VMS_DESC = """List all virtual machines across the cluster with their status and resource usage.
Example:
{"vmid": "100", "name": "ubuntu", "status": "running", "cpu": 2, "memory": 4096}"""
CREATE_VM_DESC = """Create a new virtual machine with specified configuration.
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - New VM ID number (e.g. '200', '300')
name* - VM name (e.g. 'my-new-vm', 'web-server')
cpus* - Number of CPU cores (e.g. 1, 2, 4)
memory* - Memory size in MB (e.g. 2048 for 2GB, 4096 for 4GB)
disk_size* - Disk size in GB (e.g. 10, 20, 50)
storage - Storage name (optional, will auto-detect if not specified)
ostype - OS type (optional, default: 'l26' for Linux)
Examples:
- Create VM with 1 CPU, 2GB RAM, 10GB disk: node='pve', vmid='200', name='test-vm', cpus=1, memory=2048, disk_size=10
- Create VM with 2 CPUs, 4GB RAM, 20GB disk: node='pve', vmid='201', name='web-server', cpus=2, memory=4096, disk_size=20"""
EXECUTE_VM_COMMAND_DESC = """Execute commands in a VM via QEMU guest agent.
Parameters:
node* - Host node name (e.g. 'pve1')
vmid* - VM ID number (e.g. '100')
command* - Shell command to run (e.g. 'uname -a')
Example:
{"success": true, "output": "Linux vm1 5.4.0", "exit_code": 0}"""
# VM Power Management tool descriptions
START_VM_DESC = """Start a virtual machine.
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - VM ID number (e.g. '101')
Example:
Power on VPN-Server with ID 101 on node pve"""
STOP_VM_DESC = """Stop a virtual machine (force stop).
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - VM ID number (e.g. '101')
Example:
Force stop VPN-Server with ID 101 on node pve"""
SHUTDOWN_VM_DESC = """Shutdown a virtual machine gracefully.
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - VM ID number (e.g. '101')
Example:
Gracefully shutdown VPN-Server with ID 101 on node pve"""
RESET_VM_DESC = """Reset (restart) a virtual machine.
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - VM ID number (e.g. '101')
Example:
Reset VPN-Server with ID 101 on node pve"""
DELETE_VM_DESC = """Delete/remove a virtual machine completely.
⚠️ WARNING: This operation permanently deletes the VM and all its data!
Parameters:
node* - Host node name (e.g. 'pve')
vmid* - VM ID number (e.g. '998')
force - Force deletion even if VM is running (optional, default: false)
This will permanently remove:
- VM configuration
- All virtual disks
- All snapshots
- Cannot be undone!
Example:
Delete test VM with ID 998 on node pve"""
# Container tool descriptions
GET_CONTAINERS_DESC = """List LXC containers across the cluster (or filter by node).
Parameters:
- node (optional): Node name to filter (e.g. 'pve1')
- include_stats (bool, default true): Include live CPU/memory stats
- include_raw (bool, default false): Include raw Proxmox API payloads for debugging
- format_style ('pretty'|'json', default 'pretty'): Pretty text or raw JSON list
Notes:
- Live stats from /nodes/{node}/lxc/{vmid}/status/current.
- If maxmem is 0 (unlimited), memory limit falls back to /config.memory (MiB).
- If live returns zeros, the most recent RRD sample is used as a fallback.
- Fields provided: cores (CPU cores/cpulimit), memory (MiB limit), cpu_pct, mem_bytes, maxmem_bytes, mem_pct, unlimited_memory.
"""
START_CONTAINER_DESC = """Start one or more LXC containers.
selector: '123' | 'pve1:123' | 'pve1/name' | 'name' | comma list
Example: start_container selector='pve1:101,pve2/web'
"""
STOP_CONTAINER_DESC = """Stop LXC containers. graceful=True uses shutdown; otherwise force stop.
selector: same grammar as start_container
timeout_seconds: 10 (default)
"""
RESTART_CONTAINER_DESC = """Restart LXC containers (reboot).
selector: same grammar as start_container
"""
UPDATE_CONTAINER_RESOURCES_DESC = """Update resources for one or more LXC containers.
selector: same grammar as start_container
cores: New CPU core count (optional)
memory: New memory limit in MiB (optional)
swap: New swap limit in MiB (optional)
disk_gb: Additional disk size in GiB to add (optional)
disk: Disk identifier to resize (default 'rootfs')
"""
# Storage tool descriptions
GET_STORAGE_DESC = """List storage pools across the cluster with their usage and configuration.
Example:
{"storage": "local-lvm", "type": "lvm", "used": "500GB", "total": "1TB"}"""
# Cluster tool descriptions
GET_CLUSTER_STATUS_DESC = """Get overall Proxmox cluster health and configuration status.
Example:
{"name": "proxmox", "quorum": "ok", "nodes": 3, "ha_status": "active"}"""
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/node.py:
--------------------------------------------------------------------------------
```python
"""
Node-related tools for Proxmox MCP.
This module provides tools for managing and monitoring Proxmox nodes:
- Listing all nodes in the cluster with their status
- Getting detailed node information including:
* CPU usage and configuration
* Memory utilization
* Uptime statistics
* Health status
The tools handle both basic and detailed node information retrieval,
with fallback mechanisms for partial data availability.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_NODES_DESC, GET_NODE_STATUS_DESC
class NodeTools(ProxmoxTool):
"""Tools for managing Proxmox nodes.
Provides functionality for:
- Retrieving cluster-wide node information
- Getting detailed status for specific nodes
- Monitoring node health and resources
- Handling node-specific API operations
Implements fallback mechanisms for scenarios where detailed
node information might be temporarily unavailable.
"""
def get_nodes(self) -> List[Content]:
"""List all nodes in the Proxmox cluster with detailed status.
Retrieves comprehensive information for each node including:
- Basic status (online/offline)
- Uptime statistics
- CPU configuration and count
- Memory usage and capacity
Implements a fallback mechanism that returns basic information
if detailed status retrieval fails for any node.
Returns:
List of Content objects containing formatted node information:
{
"node": "node_name",
"status": "online/offline",
"uptime": seconds,
"maxcpu": cpu_count,
"memory": {
"used": bytes,
"total": bytes
}
}
Raises:
RuntimeError: If the cluster-wide node query fails
"""
try:
result = self.proxmox.nodes.get()
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": {
# The nodes.get() API already returns memory usage
# in the "mem" field, so use that directly. The
# previous implementation subtracted this value
# from "maxmem" which actually produced the amount
# of *free* memory instead of the used memory.
"used": 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)
def get_node_status(self, node: str) -> List[Content]:
"""Get detailed status information for a specific node.
Retrieves comprehensive status information including:
- CPU usage and configuration
- Memory utilization details
- Uptime and load statistics
- Network status
- Storage health
- Running tasks and services
Args:
node: Name/ID of node to query (e.g., 'pve1', 'proxmox-node2')
Returns:
List of Content objects containing detailed node status:
{
"uptime": seconds,
"cpu": {
"usage": percentage,
"cores": count
},
"memory": {
"used": bytes,
"total": bytes,
"free": bytes
},
...additional status fields
}
Raises:
ValueError: If the specified node is not found
RuntimeError: If status retrieval fails (node offline, network issues)
"""
try:
result = self.proxmox.nodes(node).status.get()
return self._format_response((node, result), "node_status")
except Exception as e:
self._handle_error(f"get status for node {node}", e)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/components.py:
--------------------------------------------------------------------------------
```python
"""
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()}"
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/console/manager.py:
--------------------------------------------------------------------------------
```python
"""
Module for managing VM console operations.
This module provides functionality for interacting with VM consoles:
- Executing commands within VMs via QEMU guest agent
- Handling command execution lifecycle
- Managing command output and status
- Error handling and logging
The module implements a robust command execution system with:
- VM state verification
- Asynchronous command execution
- Detailed status tracking
- Comprehensive error handling
"""
import logging
from typing import Dict, Any
class VMConsoleManager:
"""Manager class for VM console operations.
Provides functionality for:
- Executing commands in VM consoles
- Managing command execution lifecycle
- Handling command output and errors
- Monitoring execution status
Uses QEMU guest agent for reliable command execution with:
- VM state verification before execution
- Asynchronous command processing
- Detailed output capture
- Comprehensive error handling
"""
def __init__(self, proxmox_api):
"""Initialize the VM console manager.
Args:
proxmox_api: Initialized ProxmoxAPI instance
"""
self.proxmox = proxmox_api
self.logger = logging.getLogger("proxmox-mcp.vm-console")
async def execute_command(self, node: str, vmid: str, command: str) -> Dict[str, Any]:
"""Execute a command in a VM's console via QEMU guest agent.
Implements a two-phase command execution process:
1. Command Initiation:
- Verifies VM exists and is running
- Initiates command execution via guest agent
- Captures command PID for tracking
2. Result Collection:
- Monitors command execution status
- Captures command output and errors
- Handles completion status
Requirements:
- VM must be running
- QEMU guest agent must be installed and active
- Command execution permissions must be enabled
Args:
node: Name of the node where VM is running (e.g., 'pve1')
vmid: ID of the VM to execute command in (e.g., '100')
command: Shell command to execute in the VM
Returns:
Dictionary containing command execution results:
{
"success": true/false,
"output": "command output",
"error": "error output if any",
"exit_code": command_exit_code
}
Raises:
ValueError: If:
- VM is not found
- VM is not running
- Guest agent is not available
RuntimeError: If:
- Command execution fails
- Unable to get command status
- API communication errors occur
"""
try:
# Verify VM exists and is running
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
if vm_status["status"] != "running":
self.logger.error(f"Failed to execute command on VM {vmid}: VM is not running")
raise ValueError(f"VM {vmid} on node {node} is not running")
# Get VM's console
self.logger.info(f"Executing command on VM {vmid} (node: {node}): {command}")
# Get the API endpoint
# Use the guest agent exec endpoint
endpoint = self.proxmox.nodes(node).qemu(vmid).agent
self.logger.debug(f"Using API endpoint: {endpoint}")
# Execute the command using two-step process
try:
# Start command execution
self.logger.info("Starting command execution...")
try:
self.logger.debug(f"Executing command via agent: {command}")
exec_result = endpoint("exec").post(command=command)
self.logger.debug(f"Raw exec response: {exec_result}")
self.logger.info(f"Command started with result: {exec_result}")
except Exception as e:
self.logger.error(f"Failed to start command: {str(e)}")
raise RuntimeError(f"Failed to start command: {str(e)}")
if 'pid' not in exec_result:
raise RuntimeError("No PID returned from command execution")
pid = exec_result['pid']
self.logger.info(f"Waiting for command completion (PID: {pid})...")
# Add a small delay to allow command to complete
import asyncio
await asyncio.sleep(1)
# Get command output using exec-status
try:
self.logger.debug(f"Getting status for PID {pid}...")
console = endpoint("exec-status").get(pid=pid)
self.logger.debug(f"Raw exec-status response: {console}")
if not console:
raise RuntimeError("No response from exec-status")
except Exception as e:
self.logger.error(f"Failed to get command status: {str(e)}")
raise RuntimeError(f"Failed to get command status: {str(e)}")
self.logger.info(f"Command completed with status: {console}")
except Exception as e:
self.logger.error(f"API call failed: {str(e)}")
raise RuntimeError(f"API call failed: {str(e)}")
self.logger.debug(f"Raw API response type: {type(console)}")
self.logger.debug(f"Raw API response: {console}")
# Handle different response structures
if isinstance(console, dict):
# Handle exec-status response format
output = console.get("out-data", "")
error = console.get("err-data", "")
exit_code = console.get("exitcode", 0)
exited = console.get("exited", 0)
if not exited:
self.logger.warning("Command may not have completed")
else:
# Some versions might return data differently
self.logger.debug(f"Unexpected response type: {type(console)}")
output = str(console)
error = ""
exit_code = 0
self.logger.debug(f"Processed output: {output}")
self.logger.debug(f"Processed error: {error}")
self.logger.debug(f"Processed exit code: {exit_code}")
self.logger.debug(f"Executed command '{command}' on VM {vmid} (node: {node})")
return {
"success": True,
"output": output,
"error": error,
"exit_code": exit_code
}
except ValueError:
# Re-raise ValueError for VM not running
raise
except Exception as e:
self.logger.error(f"Failed to execute command on VM {vmid}: {str(e)}")
if "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
raise RuntimeError(f"Failed to execute command: {str(e)}")
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/formatting/templates.py:
--------------------------------------------------------------------------------
```python
"""
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)
@staticmethod
def cluster_status(status: Dict[str, Any]) -> str:
"""Template for cluster status output.
Args:
status: Cluster status data
Returns:
Formatted cluster status string
"""
result = [f"{ProxmoxTheme.SECTIONS['configuration']} Proxmox Cluster"]
# Basic cluster info
result.extend([
"",
f" • Name: {status.get('name', 'N/A')}",
f" • Quorum: {'OK' if status.get('quorum') else 'NOT OK'}",
f" • Nodes: {status.get('nodes', 0)}",
])
# Add resource count if available
resources = status.get('resources', [])
if resources:
result.append(f" • Resources: {len(resources)}")
return "\n".join(result)
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the Proxmox MCP server.
"""
import os
import json
import pytest
from unittest.mock import Mock, patch
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from proxmox_mcp.server import ProxmoxMCPServer
@pytest.fixture
def mock_env_vars():
"""Fixture to set up test environment variables."""
env_vars = {
"PROXMOX_HOST": "test.proxmox.com",
"PROXMOX_USER": "test@pve",
"PROXMOX_TOKEN_NAME": "test_token",
"PROXMOX_TOKEN_VALUE": "test_value",
"LOG_LEVEL": "DEBUG"
}
with patch.dict(os.environ, env_vars):
yield env_vars
@pytest.fixture
def mock_proxmox():
"""Fixture to mock ProxmoxAPI."""
with patch("proxmox_mcp.core.proxmox.ProxmoxAPI") as mock:
mock.return_value.nodes.get.return_value = [
{"node": "node1", "status": "online"},
{"node": "node2", "status": "online"}
]
yield mock
@pytest.fixture
def server(mock_env_vars, mock_proxmox):
"""Fixture to create a ProxmoxMCPServer instance."""
return ProxmoxMCPServer()
def test_server_initialization(server, mock_proxmox):
"""Test server initialization with environment variables."""
assert server.config.proxmox.host == "test.proxmox.com"
assert server.config.auth.user == "test@pve"
assert server.config.auth.token_name == "test_token"
assert server.config.auth.token_value == "test_value"
assert server.config.logging.level == "DEBUG"
mock_proxmox.assert_called_once()
@pytest.mark.asyncio
async def test_list_tools(server):
"""Test listing available tools."""
tools = await server.mcp.list_tools()
assert len(tools) > 0
tool_names = [tool.name for tool in tools]
assert "get_nodes" in tool_names
assert "get_vms" in tool_names
assert "get_containers" in tool_names
assert "execute_vm_command" in tool_names
assert "update_container_resources" in tool_names
@pytest.mark.asyncio
async def test_get_nodes(server, mock_proxmox):
"""Test get_nodes tool."""
mock_proxmox.return_value.nodes.get.return_value = [
{"node": "node1", "status": "online"},
{"node": "node2", "status": "online"}
]
response = await server.mcp.call_tool("get_nodes", {})
result = json.loads(response[0].text)
assert len(result) == 2
assert result[0]["node"] == "node1"
assert result[1]["node"] == "node2"
@pytest.mark.asyncio
async def test_get_node_status_missing_parameter(server):
"""Test get_node_status tool with missing parameter."""
with pytest.raises(ToolError, match="Field required"):
await server.mcp.call_tool("get_node_status", {})
@pytest.mark.asyncio
async def test_get_node_status(server, mock_proxmox):
"""Test get_node_status tool with valid parameter."""
mock_proxmox.return_value.nodes.return_value.status.get.return_value = {
"status": "running",
"uptime": 123456
}
response = await server.mcp.call_tool("get_node_status", {"node": "node1"})
result = json.loads(response[0].text)
assert result["status"] == "running"
assert result["uptime"] == 123456
@pytest.mark.asyncio
async def test_get_vms(server, mock_proxmox):
"""Test get_vms tool."""
mock_proxmox.return_value.nodes.get.return_value = [{"node": "node1", "status": "online"}]
mock_proxmox.return_value.nodes.return_value.qemu.get.return_value = [
{"vmid": "100", "name": "vm1", "status": "running"},
{"vmid": "101", "name": "vm2", "status": "stopped"}
]
response = await server.mcp.call_tool("get_vms", {})
result = json.loads(response[0].text)
assert len(result) > 0
assert result[0]["name"] == "vm1"
assert result[1]["name"] == "vm2"
@pytest.mark.asyncio
async def test_get_containers(server, mock_proxmox):
"""Test get_containers tool."""
mock_proxmox.return_value.nodes.get.return_value = [{"node": "node1", "status": "online"}]
mock_proxmox.return_value.nodes.return_value.lxc.get.return_value = [
{"vmid": "200", "name": "container1", "status": "running"},
{"vmid": "201", "name": "container2", "status": "stopped"}
]
response = await server.mcp.call_tool("get_containers", {})
result = json.loads(response[0].text)
assert len(result) > 0
assert result[0]["name"] == "container1"
assert result[1]["name"] == "container2"
@pytest.mark.asyncio
async def test_update_container_resources(server, mock_proxmox):
"""Test update_container_resources tool."""
mock_proxmox.return_value.nodes.get.return_value = [{"node": "node1", "status": "online"}]
mock_proxmox.return_value.nodes.return_value.lxc.get.return_value = [
{"vmid": "200", "name": "container1", "status": "running"}
]
ct_api = mock_proxmox.return_value.nodes.return_value.lxc.return_value
ct_api.config.put.return_value = {}
ct_api.resize.put.return_value = {}
response = await server.mcp.call_tool(
"update_container_resources",
{"selector": "node1:200", "cores": 2, "memory": 512, "swap": 256, "disk_gb": 1},
)
result = json.loads(response[0].text)
assert result[0]["ok"] is True
ct_api.config.put.assert_called_with(cores=2, memory=512, swap=256)
ct_api.resize.put.assert_called_with(disk="rootfs", size="+1G")
@pytest.mark.asyncio
async def test_get_storage(server, mock_proxmox):
"""Test get_storage tool."""
mock_proxmox.return_value.storage.get.return_value = [
{"storage": "local", "type": "dir"},
{"storage": "ceph", "type": "rbd"}
]
response = await server.mcp.call_tool("get_storage", {})
result = json.loads(response[0].text)
assert len(result) == 2
assert result[0]["storage"] == "local"
assert result[1]["storage"] == "ceph"
@pytest.mark.asyncio
async def test_get_cluster_status(server, mock_proxmox):
"""Test get_cluster_status tool."""
mock_proxmox.return_value.cluster.status.get.return_value = {
"quorate": True,
"nodes": 2
}
response = await server.mcp.call_tool("get_cluster_status", {})
result = json.loads(response[0].text)
assert result["quorate"] is True
assert result["nodes"] == 2
@pytest.mark.asyncio
async def test_execute_vm_command_success(server, mock_proxmox):
"""Test successful VM command execution."""
# Mock VM status check
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "running"
}
# Mock command execution
mock_proxmox.return_value.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
"out": "command output",
"err": "",
"exitcode": 0
}
response = await server.mcp.call_tool("execute_vm_command", {
"node": "node1",
"vmid": "100",
"command": "ls -l"
})
result = json.loads(response[0].text)
assert result["success"] is True
assert result["output"] == "command output"
assert result["error"] == ""
assert result["exit_code"] == 0
@pytest.mark.asyncio
async def test_execute_vm_command_missing_parameters(server):
"""Test VM command execution with missing parameters."""
with pytest.raises(ToolError):
await server.mcp.call_tool("execute_vm_command", {})
@pytest.mark.asyncio
async def test_execute_vm_command_vm_not_running(server, mock_proxmox):
"""Test VM command execution when VM is not running."""
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "stopped"
}
with pytest.raises(ToolError, match="not running"):
await server.mcp.call_tool("execute_vm_command", {
"node": "node1",
"vmid": "100",
"command": "ls -l"
})
@pytest.mark.asyncio
async def test_execute_vm_command_with_error(server, mock_proxmox):
"""Test VM command execution with command error."""
# Mock VM status check
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "running"
}
# Mock command execution with error
mock_proxmox.return_value.nodes.return_value.qemu.return_value.agent.exec.post.return_value = {
"out": "",
"err": "command not found",
"exitcode": 1
}
response = await server.mcp.call_tool("execute_vm_command", {
"node": "node1",
"vmid": "100",
"command": "invalid-command"
})
result = json.loads(response[0].text)
assert result["success"] is True # API call succeeded
assert result["output"] == ""
assert result["error"] == "command not found"
assert result["exit_code"] == 1
@pytest.mark.asyncio
async def test_start_vm(server, mock_proxmox):
"""Test start_vm tool."""
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.current.get.return_value = {
"status": "stopped"
}
mock_proxmox.return_value.nodes.return_value.qemu.return_value.status.start.post.return_value = "UPID:taskid"
response = await server.mcp.call_tool("start_vm", {"node": "node1", "vmid": "100"})
assert "start initiated successfully" in response[0].text
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/server.py:
--------------------------------------------------------------------------------
```python
"""
Main server implementation for Proxmox MCP.
This module implements the core MCP server for Proxmox integration, providing:
- Configuration loading and validation
- Logging setup
- Proxmox API connection management
- MCP tool registration and routing
- Signal handling for graceful shutdown
The server exposes a set of tools for managing Proxmox resources including:
- Node management
- VM operations
- Storage management
- Cluster status monitoring
"""
import logging
import os
import sys
import signal
from typing import Optional, List, Annotated, Literal
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.tools import Tool
from mcp.types import TextContent as Content
from pydantic import Field, BaseModel
from fastapi import Body
from .config.loader import load_config
from .core.logging import setup_logging
from .core.proxmox import ProxmoxManager
from .tools.node import NodeTools
from .tools.vm import VMTools
from .tools.storage import StorageTools
from .tools.cluster import ClusterTools
from .tools.containers import ContainerTools
from .tools.definitions import (
GET_NODES_DESC,
GET_NODE_STATUS_DESC,
GET_VMS_DESC,
CREATE_VM_DESC,
EXECUTE_VM_COMMAND_DESC,
START_VM_DESC,
STOP_VM_DESC,
SHUTDOWN_VM_DESC,
RESET_VM_DESC,
DELETE_VM_DESC,
GET_CONTAINERS_DESC,
START_CONTAINER_DESC,
STOP_CONTAINER_DESC,
RESTART_CONTAINER_DESC,
UPDATE_CONTAINER_RESOURCES_DESC,
GET_STORAGE_DESC,
GET_CLUSTER_STATUS_DESC
)
class ProxmoxMCPServer:
"""Main server class for Proxmox MCP."""
def __init__(self, config_path: Optional[str] = None):
"""Initialize the server.
Args:
config_path: Path to configuration file
"""
self.config = load_config(config_path)
self.logger = setup_logging(self.config.logging)
# Initialize core components
self.proxmox_manager = ProxmoxManager(self.config.proxmox, self.config.auth)
self.proxmox = self.proxmox_manager.get_api()
# Initialize tools
self.node_tools = NodeTools(self.proxmox)
self.vm_tools = VMTools(self.proxmox)
self.storage_tools = StorageTools(self.proxmox)
self.cluster_tools = ClusterTools(self.proxmox)
self.container_tools = ContainerTools(self.proxmox)
# Initialize MCP server
self.mcp = FastMCP("ProxmoxMCP")
self._setup_tools()
def _setup_tools(self) -> None:
"""Register MCP tools with the server.
Initializes and registers all available tools with the MCP server:
- Node management tools (list nodes, get status)
- VM operation tools (list VMs, execute commands, power management)
- Storage management tools (list storage)
- Cluster tools (get cluster status)
Each tool is registered with appropriate descriptions and parameter
validation using Pydantic models.
"""
# Node tools
@self.mcp.tool(description=GET_NODES_DESC)
def get_nodes():
return self.node_tools.get_nodes()
@self.mcp.tool(description=GET_NODE_STATUS_DESC)
def get_node_status(
node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1', 'proxmox-node2')")]
):
return self.node_tools.get_node_status(node)
# VM tools
@self.mcp.tool(description=GET_VMS_DESC)
def get_vms():
return self.vm_tools.get_vms()
@self.mcp.tool(description=CREATE_VM_DESC)
def create_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="New VM ID number (e.g. '200', '300')")],
name: Annotated[str, Field(description="VM name (e.g. 'my-new-vm', 'web-server')")],
cpus: Annotated[int, Field(description="Number of CPU cores (e.g. 1, 2, 4)", ge=1, le=32)],
memory: Annotated[int, Field(description="Memory size in MB (e.g. 2048 for 2GB)", ge=512, le=131072)],
disk_size: Annotated[int, Field(description="Disk size in GB (e.g. 10, 20, 50)", ge=5, le=1000)],
storage: Annotated[Optional[str], Field(description="Storage name (optional, will auto-detect)", default=None)] = None,
ostype: Annotated[Optional[str], Field(description="OS type (optional, default: 'l26' for Linux)", default=None)] = None
):
return self.vm_tools.create_vm(node, vmid, name, cpus, memory, disk_size, storage, ostype)
@self.mcp.tool(description=EXECUTE_VM_COMMAND_DESC)
async def execute_vm_command(
node: Annotated[str, Field(description="Host node name (e.g. 'pve1', 'proxmox-node2')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '100', '101')")],
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a', 'systemctl status nginx')")]
):
return await self.vm_tools.execute_command(node, vmid, command)
# VM Power Management tools
@self.mcp.tool(description=START_VM_DESC)
def start_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '101')")]
):
return self.vm_tools.start_vm(node, vmid)
@self.mcp.tool(description=STOP_VM_DESC)
def stop_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '101')")]
):
return self.vm_tools.stop_vm(node, vmid)
@self.mcp.tool(description=SHUTDOWN_VM_DESC)
def shutdown_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '101')")]
):
return self.vm_tools.shutdown_vm(node, vmid)
@self.mcp.tool(description=RESET_VM_DESC)
def reset_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '101')")]
):
return self.vm_tools.reset_vm(node, vmid)
@self.mcp.tool(description=DELETE_VM_DESC)
def delete_vm(
node: Annotated[str, Field(description="Host node name (e.g. 'pve')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '998')")],
force: Annotated[bool, Field(description="Force deletion even if VM is running", default=False)] = False
):
return self.vm_tools.delete_vm(node, vmid, force)
# Storage tools
@self.mcp.tool(description=GET_STORAGE_DESC)
def get_storage():
return self.storage_tools.get_storage()
# Cluster tools
@self.mcp.tool(description=GET_CLUSTER_STATUS_DESC)
def get_cluster_status():
return self.cluster_tools.get_cluster_status()
# Containers (LXC)
class GetContainersPayload(BaseModel):
node: Optional[str] = Field(None, description="Optional node name (e.g. 'pve1')")
include_stats: bool = Field(True, description="Include live stats and fallbacks")
include_raw: bool = Field(False, description="Include raw status/config")
format_style: Literal["pretty", "json"] = Field(
"pretty", description="'pretty' or 'json'"
)
@self.mcp.tool(description=GET_CONTAINERS_DESC)
def get_containers(
payload: GetContainersPayload = Body(..., embed=True, description="Container query options")
):
return self.container_tools.get_containers(
node=payload.node,
include_stats=payload.include_stats,
include_raw=payload.include_raw,
format_style=payload.format_style,
)
# Container controls
@self.mcp.tool(description=START_CONTAINER_DESC)
def start_container(
selector: Annotated[str, Field(description="CT selector: '123' | 'pve1:123' | 'pve1/name' | 'name' | comma list")],
format_style: Annotated[str, Field(description="'pretty' or 'json'", pattern="^(pretty|json)$")] = "pretty",
):
return self.container_tools.start_container(selector=selector, format_style=format_style)
@self.mcp.tool(description=STOP_CONTAINER_DESC)
def stop_container(
selector: Annotated[str, Field(description="CT selector (see start_container)")],
graceful: Annotated[bool, Field(description="Graceful shutdown (True) or forced stop (False)", default=True)] = True,
timeout_seconds: Annotated[int, Field(description="Timeout for stop/shutdown", ge=1, le=600)] = 10,
format_style: Annotated[Literal["pretty","json"], Field(description="Output format")] = "pretty",
):
return self.container_tools.stop_container(
selector=selector, graceful=graceful, timeout_seconds=timeout_seconds, format_style=format_style
)
@self.mcp.tool(description=RESTART_CONTAINER_DESC)
def restart_container(
selector: Annotated[str, Field(description="CT selector (see start_container)")],
timeout_seconds: Annotated[int, Field(description="Timeout for reboot", ge=1, le=600)] = 10,
format_style: Annotated[str, Field(description="'pretty' or 'json'", pattern="^(pretty|json)$")] = "pretty",
):
return self.container_tools.restart_container(
selector=selector, timeout_seconds=timeout_seconds, format_style=format_style
)
@self.mcp.tool(description=UPDATE_CONTAINER_RESOURCES_DESC)
def update_container_resources(
selector: Annotated[str, Field(description="CT selector (see start_container)")],
cores: Annotated[Optional[int], Field(description="New CPU core count", ge=1)] = None,
memory: Annotated[Optional[int], Field(description="New memory limit in MiB", ge=16)] = None,
swap: Annotated[Optional[int], Field(description="New swap limit in MiB", ge=0)] = None,
disk_gb: Annotated[Optional[int], Field(description="Additional disk size in GiB", ge=1)] = None,
disk: Annotated[str, Field(description="Disk to resize", default="rootfs")] = "rootfs",
format_style: Annotated[Literal["pretty","json"], Field(description="Output format")] = "pretty",
):
return self.container_tools.update_container_resources(
selector=selector,
cores=cores,
memory=memory,
swap=swap,
disk_gb=disk_gb,
disk=disk,
format_style=format_style,
)
def start(self) -> None:
"""Start the MCP server.
Initializes the server with:
- Signal handlers for graceful shutdown (SIGINT, SIGTERM)
- Async runtime for handling concurrent requests
- Error handling and logging
The server runs until terminated by a signal or fatal error.
"""
import anyio
def signal_handler(signum, frame):
self.logger.info("Received signal to shutdown...")
sys.exit(0)
# Set up signal handlers
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
try:
self.logger.info("Starting MCP server...")
anyio.run(self.mcp.run_stdio_async)
except Exception as e:
self.logger.error(f"Server error: {e}")
sys.exit(1)
if __name__ == "__main__":
config_path = os.getenv("PROXMOX_MCP_CONFIG")
if not config_path:
print("PROXMOX_MCP_CONFIG environment variable must be set")
sys.exit(1)
try:
server = ProxmoxMCPServer(config_path)
server.start()
except KeyboardInterrupt:
print("\nShutting down gracefully...")
sys.exit(0)
except Exception as e:
print(f"Error: {e}")
sys.exit(1)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/vm.py:
--------------------------------------------------------------------------------
```python
"""
VM-related tools for Proxmox MCP.
This module provides tools for managing and interacting with Proxmox VMs:
- Listing all VMs across the cluster with their status
- Retrieving detailed VM information including:
* Resource allocation (CPU, memory)
* Runtime status
* Node placement
- Executing commands within VMs via QEMU guest agent
- Handling VM console operations
- VM power management (start, stop, shutdown, reset)
- VM creation with customizable specifications
The tools implement fallback mechanisms for scenarios where
detailed VM information might be temporarily unavailable.
"""
from typing import List, Optional
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_VMS_DESC, EXECUTE_VM_COMMAND_DESC
from .console.manager import VMConsoleManager
class VMTools(ProxmoxTool):
"""Tools for managing Proxmox VMs.
Provides functionality for:
- Retrieving cluster-wide VM information
- Getting detailed VM status and configuration
- Executing commands within VMs
- Managing VM console operations
- VM power management (start, stop, shutdown, reset)
- VM creation with customizable specifications
Implements fallback mechanisms for scenarios where detailed
VM information might be temporarily unavailable. Integrates
with QEMU guest agent for VM command execution.
"""
def __init__(self, proxmox_api):
"""Initialize VM tools.
Args:
proxmox_api: Initialized ProxmoxAPI instance
"""
super().__init__(proxmox_api)
self.console_manager = VMConsoleManager(proxmox_api)
def get_vms(self) -> List[Content]:
"""List all virtual machines across the cluster with detailed status.
Retrieves comprehensive information for each VM including:
- Basic identification (ID, name)
- Runtime status (running, stopped)
- Resource allocation and usage:
* CPU cores
* Memory allocation and usage
- Node placement
Implements a fallback mechanism that returns basic information
if detailed configuration retrieval fails for any VM.
Returns:
List of Content objects containing formatted VM information:
{
"vmid": "100",
"name": "vm-name",
"status": "running/stopped",
"node": "node-name",
"cpus": core_count,
"memory": {
"used": bytes,
"total": bytes
}
}
Raises:
RuntimeError: If the cluster-wide VM query fails
"""
try:
result = []
for node in self.proxmox.nodes.get():
node_name = node["node"]
vms = self.proxmox.nodes(node_name).qemu.get()
for vm in vms:
vmid = vm["vmid"]
# Get VM config for CPU cores
try:
config = self.proxmox.nodes(node_name).qemu(vmid).config.get()
result.append({
"vmid": vmid,
"name": vm["name"],
"status": vm["status"],
"node": node_name,
"cpus": config.get("cores", "N/A"),
"memory": {
"used": vm.get("mem", 0),
"total": vm.get("maxmem", 0)
}
})
except Exception:
# Fallback if can't get config
result.append({
"vmid": vmid,
"name": vm["name"],
"status": vm["status"],
"node": node_name,
"cpus": "N/A",
"memory": {
"used": vm.get("mem", 0),
"total": vm.get("maxmem", 0)
}
})
return self._format_response(result, "vms")
except Exception as e:
self._handle_error("get VMs", e)
def create_vm(self, node: str, vmid: str, name: str, cpus: int, memory: int,
disk_size: int, storage: Optional[str] = None, ostype: Optional[str] = None) -> List[Content]:
"""Create a new virtual machine with specified configuration.
Args:
node: Host node name (e.g., 'pve')
vmid: New VM ID number (e.g., '200')
name: VM name (e.g., 'my-new-vm')
cpus: Number of CPU cores (e.g., 1, 2, 4)
memory: Memory size in MB (e.g., 2048 for 2GB)
disk_size: Disk size in GB (e.g., 10, 20, 50)
storage: Storage name (e.g., 'local-lvm', 'vm-storage'). If None, will auto-detect
ostype: OS type (e.g., 'l26' for Linux, 'win10' for Windows). Default: 'l26'
Returns:
List of Content objects containing creation result
Raises:
ValueError: If VM ID already exists or invalid parameters
RuntimeError: If VM creation fails
"""
try:
# Check if VM ID already exists
try:
existing_vm = self.proxmox.nodes(node).qemu(vmid).config.get()
raise ValueError(f"VM {vmid} already exists on node {node}")
except Exception as e:
if "does not exist" not in str(e).lower():
raise e
# Get storage information
storage_list = self.proxmox.nodes(node).storage.get()
storage_info = {}
for s in storage_list:
storage_info[s["storage"]] = s
# Auto-detect storage if not specified
if storage is None:
# Prefer local-lvm for VM images first
for s in storage_list:
if s["storage"] == "local-lvm" and "images" in s.get("content", ""):
storage = s["storage"]
break
if storage is None:
# Then try vm-storage
for s in storage_list:
if s["storage"] == "vm-storage" and "images" in s.get("content", ""):
storage = s["storage"]
break
if storage is None:
# Fallback to any storage that supports images
for s in storage_list:
if "images" in s.get("content", ""):
storage = s["storage"]
break
if storage is None:
raise ValueError("No suitable storage found for VM images")
# Validate storage exists and supports images
if storage not in storage_info:
raise ValueError(f"Storage '{storage}' not found on node {node}")
if "images" not in storage_info[storage].get("content", ""):
raise ValueError(f"Storage '{storage}' does not support VM images")
# Determine appropriate disk format based on storage type
storage_type = storage_info[storage]["type"]
if storage_type in ["lvm", "lvmthin"]:
# LVM storages use raw format and no cloudinit
disk_format = "raw"
vm_config_storage = {
"scsi0": f"{storage}:{disk_size},format={disk_format}",
}
elif storage_type in ["dir", "nfs", "cifs"]:
# File-based storages can use qcow2
disk_format = "qcow2"
vm_config_storage = {
"scsi0": f"{storage}:{disk_size},format={disk_format}",
"ide2": f"{storage}:cloudinit",
}
else:
# Default to raw for unknown storage types
disk_format = "raw"
vm_config_storage = {
"scsi0": f"{storage}:{disk_size},format={disk_format}",
}
# Set default OS type
if ostype is None:
ostype = "l26" # Linux 2.6+ kernel
# Prepare VM configuration
vm_config = {
"vmid": vmid,
"name": name,
"cores": cpus,
"memory": memory,
"ostype": ostype,
"scsihw": "virtio-scsi-pci",
"boot": "order=scsi0",
"agent": "1", # Enable QEMU guest agent
"vga": "std",
"net0": "virtio,bridge=vmbr0",
}
# Add storage configuration
vm_config.update(vm_config_storage)
# Create the VM
task_result = self.proxmox.nodes(node).qemu.create(**vm_config)
cloudinit_note = ""
if storage_type in ["lvm", "lvmthin"]:
cloudinit_note = "\n ⚠️ Note: LVM storage doesn't support cloud-init image"
result_text = f"""🎉 VM {vmid} created successfully!
📋 VM Configuration:
• Name: {name}
• Node: {node}
• VM ID: {vmid}
• CPU Cores: {cpus}
• Memory: {memory} MB ({memory/1024:.1f} GB)
• Disk: {disk_size} GB ({storage}, {disk_format} format)
• Storage Type: {storage_type}
• OS Type: {ostype}
• Network: virtio (bridge=vmbr0)
• QEMU Agent: Enabled{cloudinit_note}
🔧 Task ID: {task_result}
💡 Next steps:
1. Upload an ISO to install the operating system
2. Start the VM using start_vm tool
3. Access the console to complete OS installation"""
return [Content(type="text", text=result_text)]
except ValueError as e:
raise e
except Exception as e:
self._handle_error(f"create VM {vmid}", e)
def start_vm(self, node: str, vmid: str) -> List[Content]:
"""Start a virtual machine.
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
Returns:
List of Content objects containing operation result
Raises:
ValueError: If VM is not found
RuntimeError: If start operation fails
"""
try:
# Check if VM exists and get current status
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
current_status = vm_status.get("status")
if current_status == "running":
result_text = f"🟢 VM {vmid} is already running"
else:
# Start the VM
task_result = self.proxmox.nodes(node).qemu(vmid).status.start.post()
result_text = f"🚀 VM {vmid} start initiated successfully\nTask ID: {task_result}"
return [Content(type="text", text=result_text)]
except Exception as e:
if "does not exist" in str(e).lower() or "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
self._handle_error(f"start VM {vmid}", e)
def stop_vm(self, node: str, vmid: str) -> List[Content]:
"""Stop a virtual machine (force stop).
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
Returns:
List of Content objects containing operation result
Raises:
ValueError: If VM is not found
RuntimeError: If stop operation fails
"""
try:
# Check if VM exists and get current status
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
current_status = vm_status.get("status")
if current_status == "stopped":
result_text = f"🔴 VM {vmid} is already stopped"
else:
# Stop the VM
task_result = self.proxmox.nodes(node).qemu(vmid).status.stop.post()
result_text = f"🛑 VM {vmid} stop initiated successfully\nTask ID: {task_result}"
return [Content(type="text", text=result_text)]
except Exception as e:
if "does not exist" in str(e).lower() or "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
self._handle_error(f"stop VM {vmid}", e)
def shutdown_vm(self, node: str, vmid: str) -> List[Content]:
"""Shutdown a virtual machine gracefully.
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
Returns:
List of Content objects containing operation result
Raises:
ValueError: If VM is not found
RuntimeError: If shutdown operation fails
"""
try:
# Check if VM exists and get current status
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
current_status = vm_status.get("status")
if current_status == "stopped":
result_text = f"🔴 VM {vmid} is already stopped"
else:
# Shutdown the VM gracefully
task_result = self.proxmox.nodes(node).qemu(vmid).status.shutdown.post()
result_text = f"💤 VM {vmid} graceful shutdown initiated\nTask ID: {task_result}"
return [Content(type="text", text=result_text)]
except Exception as e:
if "does not exist" in str(e).lower() or "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
self._handle_error(f"shutdown VM {vmid}", e)
def reset_vm(self, node: str, vmid: str) -> List[Content]:
"""Reset (restart) a virtual machine.
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
Returns:
List of Content objects containing operation result
Raises:
ValueError: If VM is not found
RuntimeError: If reset operation fails
"""
try:
# Check if VM exists and get current status
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
current_status = vm_status.get("status")
if current_status == "stopped":
result_text = f"⚠️ Cannot reset VM {vmid}: VM is currently stopped\nUse start_vm to start it first"
else:
# Reset the VM
task_result = self.proxmox.nodes(node).qemu(vmid).status.reset.post()
result_text = f"🔄 VM {vmid} reset initiated successfully\nTask ID: {task_result}"
return [Content(type="text", text=result_text)]
except Exception as e:
if "does not exist" in str(e).lower() or "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
self._handle_error(f"reset VM {vmid}", e)
async def execute_command(self, node: str, vmid: str, command: str) -> List[Content]:
"""Execute a command in a VM via QEMU guest agent.
Uses the QEMU guest agent to execute commands within a running VM.
Requires:
- VM must be running
- QEMU guest agent must be installed and running in the VM
- Command execution permissions must be enabled
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
command: Shell command to run (e.g., 'uname -a', 'systemctl status nginx')
Returns:
List of Content objects containing formatted command output:
{
"success": true/false,
"output": "command output",
"error": "error message if any"
}
Raises:
ValueError: If VM is not found, not running, or guest agent is not available
RuntimeError: If command execution fails due to permissions or other issues
"""
try:
result = await self.console_manager.execute_command(node, vmid, command)
# 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)
def delete_vm(self, node: str, vmid: str, force: bool = False) -> List[Content]:
"""Delete/remove a virtual machine completely.
This will permanently delete the VM and all its associated data including:
- VM configuration
- Virtual disks
- Snapshots
WARNING: This operation cannot be undone!
Args:
node: Host node name (e.g., 'pve1', 'proxmox-node2')
vmid: VM ID number (e.g., '100', '101')
force: Force deletion even if VM is running (will stop first)
Returns:
List of Content objects containing deletion result
Raises:
ValueError: If VM is not found or is running and force=False
RuntimeError: If deletion fails
"""
try:
# Check if VM exists and get current status
try:
vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
current_status = vm_status.get("status")
vm_name = vm_status.get("name", f"VM-{vmid}")
except Exception as e:
if "does not exist" in str(e).lower() or "not found" in str(e).lower():
raise ValueError(f"VM {vmid} not found on node {node}")
raise e
# Check if VM is running
if current_status == "running":
if not force:
raise ValueError(f"VM {vmid} ({vm_name}) is currently running. "
f"Please stop it first or use force=True to stop and delete.")
else:
# Force stop the VM first
self.proxmox.nodes(node).qemu(vmid).status.stop.post()
result_text = f"🛑 Stopping VM {vmid} ({vm_name}) before deletion...\n"
else:
result_text = f"🗑️ Deleting VM {vmid} ({vm_name})...\n"
# Delete the VM
task_result = self.proxmox.nodes(node).qemu(vmid).delete()
result_text += f"""🗑️ VM {vmid} ({vm_name}) deletion initiated successfully!
⚠️ WARNING: This operation will permanently remove:
• VM configuration
• All virtual disks
• All snapshots
• Cannot be undone!
🔧 Task ID: {task_result}
✅ VM {vmid} ({vm_name}) is being deleted from node {node}"""
return [Content(type="text", text=result_text)]
except ValueError as e:
raise e
except Exception as e:
self._handle_error(f"delete VM {vmid}", e)
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/containers.py:
--------------------------------------------------------------------------------
```python
from typing import List, Dict, Optional, Tuple, Any, Union
import json
from mcp.types import TextContent as Content
from .base import ProxmoxTool
def _b2h(n: Union[int, float, str]) -> str:
"""bytes -> human (binary units)."""
try:
n = float(n)
except Exception:
return "0.00 B"
units = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
i = 0
while n >= 1024.0 and i < len(units) - 1:
n /= 1024.0
i += 1
return f"{n:.2f} {units[i]}"
# The rest of the helpers were preserved from your original file; no changes needed
def _get(d: Any, key: str, default: Any = None) -> Any:
"""dict.get with None guard."""
if isinstance(d, dict):
return d.get(key, default)
return default
def _as_dict(maybe: Any) -> Dict:
"""Return dict; unwrap {'data': dict}; else {}."""
if isinstance(maybe, dict):
data = maybe.get("data")
if isinstance(data, dict):
return data
return maybe
return {}
def _as_list(maybe: Any) -> List:
"""Return list; unwrap {'data': list}; else []."""
if isinstance(maybe, list):
return maybe
if isinstance(maybe, dict):
data = maybe.get("data")
if isinstance(data, list):
return data
return []
class ContainerTools(ProxmoxTool):
"""
LXC container tools for Proxmox MCP.
- Lists containers cluster-wide (or by node)
- Live stats via /status/current
- Limit fallback via /config (memory MiB, cores/cpulimit)
- RRD fallback when live returns zeros
- Pretty output rendered here; JSON path is raw & sanitized
"""
# ---------- error / output ----------
def _json_fmt(self, data: Any) -> List[Content]:
"""Return raw JSON string (never touch project formatters)."""
return [Content(type="text", text=json.dumps(data, indent=2, sort_keys=True))]
def _err(self, action: str, e: Exception) -> List[Content]:
if hasattr(self, "handle_error"):
return self.handle_error(e, action) # type: ignore[attr-defined]
if hasattr(self, "_handle_error"):
return self._handle_error(action, e) # type: ignore[attr-defined]
return [Content(type="text", text=json.dumps({"error": str(e), "action": action}))]
# ---------- helpers ----------
def _list_ct_pairs(self, node: Optional[str]) -> List[Tuple[str, Dict]]:
"""Yield (node_name, ct_dict). Coerce odd shapes into dicts with vmid."""
out: List[Tuple[str, Dict]] = []
if node:
raw = self.proxmox.nodes(node).lxc.get()
for it in _as_list(raw):
if isinstance(it, dict):
out.append((node, it))
else:
try:
vmid = int(it)
out.append((node, {"vmid": vmid}))
except Exception:
continue
else:
nodes = _as_list(self.proxmox.nodes.get())
for n in nodes:
nname = _get(n, "node")
if not nname:
continue
raw = self.proxmox.nodes(nname).lxc.get()
for it in _as_list(raw):
if isinstance(it, dict):
out.append((nname, it))
else:
try:
vmid = int(it)
out.append((nname, {"vmid": vmid}))
except Exception:
continue
return out
def _rrd_last(self, node: str, vmid: int) -> Tuple[Optional[float], Optional[int], Optional[int]]:
"""Return (cpu_pct, mem_bytes, maxmem_bytes) from the most recent RRD sample."""
try:
rrd = _as_list(self.proxmox.nodes(node).lxc(vmid).rrddata.get(timeframe="hour", ds="cpu,mem,maxmem"))
if not rrd or not isinstance(rrd[-1], dict):
return None, None, None
last = rrd[-1]
# Proxmox RRD cpu is fraction already (0..1). Convert to percent.
cpu_pct = float(_get(last, "cpu", 0.0) or 0.0) * 100.0
mem_bytes = int(_get(last, "mem", 0) or 0)
maxmem_bytes = int(_get(last, "maxmem", 0) or 0)
return cpu_pct, mem_bytes, maxmem_bytes
except Exception:
return None, None, None
def _status_and_config(self, node: str, vmid: int) -> Tuple[Dict, Dict]:
"""Return (status_current_dict, config_dict)."""
raw_status: Dict = {}
raw_config: Dict = {}
try:
raw_status = _as_dict(self.proxmox.nodes(node).lxc(vmid).status.current.get())
except Exception:
raw_status = {}
try:
raw_config = _as_dict(self.proxmox.nodes(node).lxc(vmid).config.get())
except Exception:
raw_config = {}
return raw_status, raw_config
def _render_pretty(self, rows: List[Dict]) -> List[Content]:
lines: List[str] = ["📦 Containers", ""]
for r in rows:
name = r.get("name") or f"ct-{r.get('vmid')}"
vmid = r.get("vmid")
status = (r.get("status") or "").upper()
node = r.get("node") or "?"
cores = r.get("cores")
cpu_pct = r.get("cpu_pct", 0.0)
mem_bytes = int(r.get("mem_bytes") or 0)
maxmem_bytes = int(r.get("maxmem_bytes") or 0)
mem_pct = r.get("mem_pct")
unlimited = bool(r.get("unlimited_memory", False))
lines.append(f"📦 {name} (ID: {vmid})")
lines.append(f" • Status: {status}")
lines.append(f" • Node: {node}")
lines.append(f" • CPU: {cpu_pct:.1f}%")
lines.append(f" • CPU Cores: {cores if cores is not None else 'N/A'}")
if unlimited:
lines.append(f" • Memory: {_b2h(mem_bytes)} (unlimited)")
else:
if maxmem_bytes > 0:
pct_str = f" ({mem_pct:.1f}%)" if isinstance(mem_pct, (int, float)) else ""
lines.append(f" • Memory: {_b2h(mem_bytes)} / {_b2h(maxmem_bytes)}{pct_str}")
else:
lines.append(f" • Memory: {_b2h(mem_bytes)} / 0.00 B")
lines.append("")
return [Content(type="text", text="\n".join(lines).rstrip())]
# ---------- tool ----------
def get_containers(
self,
node: Optional[str] = None,
include_stats: bool = True,
include_raw: bool = False,
format_style: str = "pretty",
) -> List[Content]:
"""
List containers cluster-wide or by node.
- `include_stats=True` fetches live CPU/mem from /status/current
- RRD fallback is used if live returns zeros
- `format_style='json'` returns raw JSON list (sanitized)
- `format_style='pretty'` renders a human-friendly table
"""
try:
pairs = self._list_ct_pairs(node)
rows: List[Dict] = []
for nname, ct in pairs:
vmid_val = _get(ct, "vmid")
vmid_int: Optional[int] = None
try:
if vmid_val is not None:
vmid_int = int(vmid_val)
except Exception:
vmid_int = None
rec: Dict = {
"vmid": str(vmid_val) if vmid_val is not None else None,
"name": _get(ct, "name") or _get(ct, "hostname") or (f"ct-{vmid_val}" if vmid_val is not None else "ct-?"),
"node": nname,
"status": _get(ct, "status"),
}
if include_stats and vmid_int is not None:
raw_status, raw_config = self._status_and_config(nname, vmid_int)
cpu_frac = float(_get(raw_status, "cpu", 0.0) or 0.0)
cpu_pct = round(cpu_frac * 100.0, 2)
mem_bytes = int(_get(raw_status, "mem", 0) or 0)
maxmem_bytes = int(_get(raw_status, "maxmem", 0) or 0)
memory_mib = 0
cores: Optional[Union[int, float]] = None
unlimited_memory = False
try:
cfg_mem = _get(raw_config, "memory")
if cfg_mem is None:
cfg_mem = _get(raw_config, "ram")
if cfg_mem is None:
cfg_mem = _get(raw_config, "maxmem")
if cfg_mem is None:
cfg_mem = _get(raw_config, "memoryMiB")
if cfg_mem is not None:
try:
memory_mib = int(cfg_mem)
except Exception:
memory_mib = 0
else:
memory_mib = 0
unlimited_memory = bool(_get(raw_config, "swap", 0) == 0 and memory_mib == 0)
cfg_cores = _get(raw_config, "cores")
cfg_cpulimit = _get(raw_config, "cpulimit")
if cfg_cores is not None:
cores = int(cfg_cores)
elif cfg_cpulimit is not None and float(cfg_cpulimit) > 0:
cores = float(cfg_cpulimit)
except Exception:
cores = None
# --- NEW: fallbacks for stopped / missing maxmem ---
status_str = str(_get(raw_status, "status") or _get(ct, "status") or "").lower()
if status_str == "stopped":
try:
mem_bytes = 0
except Exception:
mem_bytes = 0
if (not maxmem_bytes or int(maxmem_bytes) == 0) and memory_mib and int(memory_mib) > 0:
try:
maxmem_bytes = int(memory_mib) * 1024 * 1024
except Exception:
maxmem_bytes = 0
# RRD fallback if zeros
if (mem_bytes == 0) or (maxmem_bytes == 0) or (cpu_pct == 0.0):
rrd_cpu, rrd_mem, rrd_maxmem = self._rrd_last(nname, vmid_int)
if cpu_pct == 0.0 and rrd_cpu is not None:
cpu_pct = rrd_cpu
if mem_bytes == 0 and rrd_mem is not None:
mem_bytes = rrd_mem
if maxmem_bytes == 0 and rrd_maxmem:
maxmem_bytes = rrd_maxmem
if memory_mib == 0:
try:
memory_mib = int(round(maxmem_bytes / (1024 * 1024)))
except Exception:
memory_mib = 0
rec.update({
"cores": cores,
"memory": memory_mib,
"cpu_pct": cpu_pct,
"mem_bytes": mem_bytes,
"maxmem_bytes": maxmem_bytes,
"mem_pct": (
round((mem_bytes / maxmem_bytes * 100.0), 2)
if (maxmem_bytes and maxmem_bytes > 0)
else None
),
"unlimited_memory": unlimited_memory,
})
# For PRETTY only: allow raw blobs to be attached if requested.
if include_raw and format_style != "json":
rec["raw_status"] = raw_status
rec["raw_config"] = raw_config
rows.append(rec)
if format_style == "json":
# JSON path must be immune to any formatter assumptions; no raw payloads.
return self._json_fmt(rows)
return self._render_pretty(rows)
except Exception as e:
return self._err("Failed to list containers", e)
# ---------- target resolution for control ops ----------
def _resolve_targets(self, selector: str) -> List[Tuple[str, int, str]]:
"""
Turn a selector string into a list of (node, vmid, label).
Supports:
- '123' (vmid across cluster)
- 'pve1:123' (node:vmid)
- 'pve1/name' (node/name)
- 'name' (by name/hostname across the cluster)
- comma-separated list of any of the above
"""
if not selector:
return []
tokens = [t.strip() for t in selector.split(",") if t.strip()]
inventory: List[Tuple[str, Dict[str, Any]]] = self._list_ct_pairs(node=None)
resolved: List[Tuple[str, int, str]] = []
for tok in tokens:
if ":" in tok and "/" not in tok:
node, vmid_s = tok.split(":", 1)
try:
vmid = int(vmid_s)
except Exception:
continue
for n, ct in inventory:
if n == node and int(_get(ct, "vmid", -1)) == vmid:
label = _get(ct, "name") or _get(ct, "hostname") or f"ct-{vmid}"
resolved.append((node, vmid, label))
break
continue
if "/" in tok and ":" not in tok:
node, name = tok.split("/", 1)
name = name.strip()
for n, ct in inventory:
if n == node and (_get(ct, "name") == name or _get(ct, "hostname") == name):
vmid = int(_get(ct, "vmid", -1))
if vmid >= 0:
resolved.append((node, vmid, name))
continue
if tok.isdigit():
vmid = int(tok)
for n, ct in inventory:
if int(_get(ct, "vmid", -1)) == vmid:
label = _get(ct, "name") or _get(ct, "hostname") or f"ct-{vmid}"
resolved.append((n, vmid, label))
continue
name = tok
for n, ct in inventory:
if _get(ct, "name") == name or _get(ct, "hostname") == name:
vmid = int(_get(ct, "vmid", -1))
if vmid >= 0:
resolved.append((n, vmid, name))
uniq = {}
for n, v, lbl in resolved:
uniq[(n, v)] = lbl
return [(n, v, uniq[(n, v)]) for (n, v) in uniq.keys()]
def _render_action_result(self, title: str, results: List[Dict[str, Any]]) -> List[Content]:
"""Pretty-print an action result; JSON stays raw."""
lines = [f"📦 {title}", ""]
for r in results:
status = "✅ OK" if r.get("ok") else "❌ FAIL"
node = r.get("node")
vmid = r.get("vmid")
name = r.get("name") or f"ct-{vmid}"
msg = r.get("message") or r.get("error") or ""
lines.append(f"{status} {name} (ID: {vmid}, node: {node}) {('- ' + str(msg)) if msg else ''}")
return [Content(type="text", text="\n".join(lines).rstrip())]
# ---------- container control tools ----------
def start_container(self, selector: str, format_style: str = "pretty") -> List[Content]:
"""
Start LXC containers matching `selector`.
selector examples: '123', 'pve1:123', 'pve1/name', 'name', 'pve1:101,pve2/web'
"""
try:
targets = self._resolve_targets(selector)
if not targets:
return self._err("No containers matched the selector", ValueError(selector))
results: List[Dict[str, Any]] = []
for node, vmid, label in targets:
try:
resp = self.proxmox.nodes(node).lxc(vmid).status.start.post()
results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
except Exception as e:
results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
if format_style == "json":
return self._json_fmt(results)
return self._render_action_result("Start Containers", results)
except Exception as e:
return self._err("Failed to start container(s)", e)
def stop_container(self, selector: str, graceful: bool = True, timeout_seconds: int = 10,
format_style: str = "pretty") -> List[Content]:
"""
Stop LXC containers.
graceful=True → POST .../status/shutdown (graceful stop)
graceful=False → POST .../status/stop (force stop)
"""
try:
targets = self._resolve_targets(selector)
if not targets:
return self._err("No containers matched the selector", ValueError(selector))
results: List[Dict[str, Any]] = []
for node, vmid, label in targets:
try:
if graceful:
resp = self.proxmox.nodes(node).lxc(vmid).status.shutdown.post(timeout=timeout_seconds)
else:
resp = self.proxmox.nodes(node).lxc(vmid).status.stop.post()
results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
except Exception as e:
results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
if format_style == "json":
return self._json_fmt(results)
return self._render_action_result("Stop Containers", results)
except Exception as e:
return self._err("Failed to stop container(s)", e)
def restart_container(self, selector: str, timeout_seconds: int = 10,
format_style: str = "pretty") -> List[Content]:
"""
Restart LXC containers via POST .../status/reboot.
"""
try:
targets = self._resolve_targets(selector)
if not targets:
return self._err("No containers matched the selector", ValueError(selector))
results: List[Dict[str, Any]] = []
for node, vmid, label in targets:
try:
resp = self.proxmox.nodes(node).lxc(vmid).status.reboot.post()
results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
except Exception as e:
results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
if format_style == "json":
return self._json_fmt(results)
return self._render_action_result("Restart Containers", results)
except Exception as e:
return self._err("Failed to restart container(s)", e)
def update_container_resources(
self,
selector: str,
cores: Optional[int] = None,
memory: Optional[int] = None,
swap: Optional[int] = None,
disk_gb: Optional[int] = None,
disk: str = "rootfs",
format_style: str = "pretty",
) -> List[Content]:
"""Update container CPU/memory/swap limits and/or extend disk size.
Parameters:
selector: Container selector (same grammar as start_container)
cores: New CPU core count
memory: New memory limit in MiB
swap: New swap limit in MiB
disk_gb: Additional disk size to add in GiB
disk: Disk identifier to resize (default 'rootfs')
format_style: Output format ('pretty' or 'json')
"""
try:
targets = self._resolve_targets(selector)
if not targets:
return self._err("No containers matched the selector", ValueError(selector))
results: List[Dict[str, Any]] = []
for node, vmid, label in targets:
rec: Dict[str, Any] = {"ok": True, "node": node, "vmid": vmid, "name": label}
changes: List[str] = []
try:
update_params: Dict[str, Any] = {}
if cores is not None:
update_params["cores"] = cores
changes.append(f"cores={cores}")
if memory is not None:
update_params["memory"] = memory
changes.append(f"memory={memory}MiB")
if swap is not None:
update_params["swap"] = swap
changes.append(f"swap={swap}MiB")
if update_params:
self.proxmox.nodes(node).lxc(vmid).config.put(**update_params)
if disk_gb is not None:
size_str = f"+{disk_gb}G"
# Use PUT for disk resize - some Proxmox versions reject POST
self.proxmox.nodes(node).lxc(vmid).resize.put(disk=disk, size=size_str)
changes.append(f"{disk}+={disk_gb}G")
rec["message"] = ", ".join(changes) if changes else "no changes"
except Exception as e:
rec["ok"] = False
rec["error"] = str(e)
results.append(rec)
if format_style == "json":
return self._json_fmt(results)
return self._render_action_result("Update Container Resources", results)
except Exception as e:
return self._err("Failed to update container(s)", e)
```