From 325a4dffdfb794ec574986a4e12a1002d410e5af Mon Sep 17 00:00:00 2001 From: mineracks <134782215+mineracks@users.noreply.github.com> Date: Sun, 26 Apr 2026 09:34:29 +1000 Subject: [PATCH] =?UTF-8?q?feat:=20initial=20Vultr=20MCP=20server=20?= =?UTF-8?q?=E2=80=94=2016=20tools=20for=20VPS=20management?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 1 + README.md | 88 +++++++++ requirements.txt | 2 + server.py | 451 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 542 insertions(+) create mode 100644 .env.example create mode 100644 README.md create mode 100644 requirements.txt create mode 100644 server.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..1e051e1 --- /dev/null +++ b/.env.example @@ -0,0 +1 @@ +VULTR_API_KEY=your_api_key_here diff --git a/README.md b/README.md new file mode 100644 index 0000000..8e76830 --- /dev/null +++ b/README.md @@ -0,0 +1,88 @@ +# Vultr MCP Server + +A [Model Context Protocol](https://modelcontextprotocol.io/) server for the [Vultr](https://www.vultr.com/) VPS API. + +## Tools + +| Tool | Description | +|------|-------------| +| `list_regions` | List all available Vultr regions | +| `list_plans` | List plans, optionally filtered by region | +| `list_os` | List available OS images | +| `list_instances` | List all instances on the account | +| `get_instance` | Get details for a specific instance | +| `create_instance` | Create a new instance | +| `destroy_instance` | Permanently destroy an instance | +| `start_instance` | Power on an instance | +| `stop_instance` | Power off an instance | +| `reboot_instance` | Reboot an instance | +| `list_ssh_keys` | List SSH keys on the account | +| `create_ssh_key` | Add a new SSH key | +| `list_firewalls` | List firewall groups | +| `get_firewall` | Get rules for a firewall group | +| `create_firewall` | Create a new firewall group | +| `create_firewall_rule` | Add a rule to a firewall group | +| `get_account` | Get account info and current balance | + +## Setup + +### 1. Install dependencies + +```bash +cd /tmp/vultr-mcp +pip install -r requirements.txt +``` + +### 2. Set the API key + +Get your API key from the [Vultr control panel](https://my.vultr.com/settings/#settingsapi). + +```bash +export VULTR_API_KEY=your_api_key_here +``` + +Or copy `.env.example` to `.env` and fill it in, then source it: + +```bash +cp .env.example .env +# edit .env +source .env +``` + +### 3. Run the server + +```bash +python server.py +``` + +## Adding to Claude Code + +Add the following to your Claude Code MCP config (typically `~/.claude/claude_desktop_config.json` or via `claude mcp add`): + +```json +{ + "mcpServers": { + "vultr": { + "command": "python", + "args": ["/tmp/vultr-mcp/server.py"], + "env": { + "VULTR_API_KEY": "your_api_key_here" + } + } + } +} +``` + +Or using the Claude Code CLI: + +```bash +claude mcp add vultr \ + --command "python /tmp/vultr-mcp/server.py" \ + --env VULTR_API_KEY=your_api_key_here +``` + +## Notes + +- `destroy_instance` is irreversible — double-check the instance ID before calling it. +- Newly created instances may show `pending` as their main IP for a minute or two while they provision. +- Firewall rules default to inbound IPv4 (`ip_type: v4`, `direction: in`). Use `0.0.0.0` with `subnet_size: 0` to match all sources. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9714bd9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +mcp>=1.0.0 +httpx>=0.27.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..e4f00ac --- /dev/null +++ b/server.py @@ -0,0 +1,451 @@ +import os +import httpx +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("vultr", description="Vultr VPS hosting API") + +VULTR_API_BASE = "https://api.vultr.com/v2" + + +def get_headers(): + api_key = os.environ.get("VULTR_API_KEY") + if not api_key: + raise ValueError("VULTR_API_KEY environment variable is not set") + return {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + + +async def api_get(path: str, params: dict = None) -> dict: + async with httpx.AsyncClient() as client: + resp = await client.get(f"{VULTR_API_BASE}{path}", headers=get_headers(), params=params, timeout=30) + resp.raise_for_status() + return resp.json() + + +async def api_post(path: str, body: dict = None) -> dict: + async with httpx.AsyncClient() as client: + resp = await client.post(f"{VULTR_API_BASE}{path}", headers=get_headers(), json=body or {}, timeout=30) + resp.raise_for_status() + return resp.json() if resp.content else {} + + +async def api_delete(path: str) -> str: + async with httpx.AsyncClient() as client: + resp = await client.delete(f"{VULTR_API_BASE}{path}", headers=get_headers(), timeout=30) + resp.raise_for_status() + return "Success" + + +# --------------------------------------------------------------------------- +# Regions & Plans +# --------------------------------------------------------------------------- + +@mcp.tool() +async def list_regions() -> str: + """List all available Vultr regions.""" + try: + data = await api_get("/regions") + regions = data.get("regions", []) + lines = [f"{'ID':<15} {'City':<20} {'Country':<8} {'Continent':<15} {'Options'}"] + lines.append("-" * 80) + for r in regions: + opts = ", ".join(r.get("options", [])) + lines.append(f"{r['id']:<15} {r['city']:<20} {r['country']:<8} {r['continent']:<15} {opts}") + return f"Regions ({len(regions)} total):\n" + "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def list_plans(region: str = None) -> str: + """List available plans, optionally filtered by region availability.""" + try: + params = {} + if region: + params["region"] = region + data = await api_get("/plans", params=params if params else None) + plans = data.get("plans", []) + lines = [f"{'ID':<30} {'vCPUs':>6} {'RAM MB':>8} {'Disk GB':>8} {'BW GB':>8} {'$/mo':>8} {'Type':<12}"] + lines.append("-" * 90) + for p in plans: + lines.append( + f"{p['id']:<30} {p['vcpu_count']:>6} {p['ram']:>8} {p['disk']:>8} " + f"{p['bandwidth']:>8} {p['monthly_cost']:>8.2f} {p.get('type', ''):<12}" + ) + header = f"Plans ({len(plans)} total)" + if region: + header += f" available in {region}" + return header + ":\n" + "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +# --------------------------------------------------------------------------- +# OS Images +# --------------------------------------------------------------------------- + +@mcp.tool() +async def list_os() -> str: + """List available OS images.""" + try: + data = await api_get("/os") + images = data.get("os", []) + lines = [f"{'ID':>6} {'Name':<40} {'Family':<15} {'Arch'}"] + lines.append("-" * 70) + for img in images: + lines.append(f"{img['id']:>6} {img['name']:<40} {img['family']:<15} {img['arch']}") + return f"OS Images ({len(images)} total):\n" + "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +# --------------------------------------------------------------------------- +# Instances +# --------------------------------------------------------------------------- + +@mcp.tool() +async def list_instances() -> str: + """List all Vultr instances.""" + try: + data = await api_get("/instances") + instances = data.get("instances", []) + if not instances: + return "No instances found." + lines = [] + for inst in instances: + lines.append( + f"ID: {inst['id']}\n" + f"Label: {inst.get('label', '(none)')}\n" + f"Hostname: {inst.get('hostname', '(none)')}\n" + f"Status: {inst.get('status')} Power: {inst.get('power_status')}\n" + f"Region: {inst.get('region')} Plan: {inst.get('plan')}\n" + f"IP: {inst.get('main_ip')} IPv6: {inst.get('v6_main_ip', 'n/a')}\n" + f"OS: {inst.get('os')}\n" + f"vCPUs: {inst.get('vcpu_count')} RAM: {inst.get('ram')} MB Disk: {inst.get('disk')} GB\n" + ) + return f"Instances ({len(instances)} total):\n\n" + "\n---\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def get_instance(instance_id: str) -> str: + """Get details for a specific instance.""" + try: + data = await api_get(f"/instances/{instance_id}") + inst = data.get("instance", {}) + return ( + f"ID: {inst['id']}\n" + f"Label: {inst.get('label', '(none)')}\n" + f"Hostname: {inst.get('hostname', '(none)')}\n" + f"Status: {inst.get('status')}\n" + f"Power status: {inst.get('power_status')}\n" + f"Server state: {inst.get('server_status')}\n" + f"Region: {inst.get('region')}\n" + f"Plan: {inst.get('plan')}\n" + f"Main IP: {inst.get('main_ip')}\n" + f"IPv6: {inst.get('v6_main_ip', 'n/a')}\n" + f"OS: {inst.get('os')} (os_id: {inst.get('os_id')})\n" + f"vCPUs: {inst.get('vcpu_count')}\n" + f"RAM: {inst.get('ram')} MB\n" + f"Disk: {inst.get('disk')} GB\n" + f"Bandwidth: {inst.get('allowed_bandwidth')} GB\n" + f"Created: {inst.get('date_created')}\n" + f"Tags: {', '.join(inst.get('tags', [])) or '(none)'}\n" + f"Firewall: {inst.get('firewall_group_id') or '(none)'}\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def create_instance(region: str, plan: str, os_id: int, label: str, hostname: str, ssh_key_ids: list = None) -> str: + """Create a new Vultr instance.""" + try: + body = { + "region": region, + "plan": plan, + "os_id": os_id, + "label": label, + "hostname": hostname, + } + if ssh_key_ids: + body["sshkey_id"] = ssh_key_ids + data = await api_post("/instances", body) + inst = data.get("instance", {}) + return ( + f"Instance created successfully.\n" + f"ID: {inst['id']}\n" + f"Label: {inst.get('label')}\n" + f"Hostname: {inst.get('hostname')}\n" + f"Status: {inst.get('status')}\n" + f"Region: {inst.get('region')}\n" + f"Plan: {inst.get('plan')}\n" + f"Main IP: {inst.get('main_ip')} (may be pending — check back shortly)\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def destroy_instance(instance_id: str) -> str: + """Permanently destroy a Vultr instance. This cannot be undone.""" + try: + await api_delete(f"/instances/{instance_id}") + return f"Instance {instance_id} has been destroyed." + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def start_instance(instance_id: str) -> str: + """Start (power on) a Vultr instance.""" + try: + await api_post(f"/instances/{instance_id}/start") + return f"Start command sent to instance {instance_id}." + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def stop_instance(instance_id: str) -> str: + """Stop (power off) a Vultr instance.""" + try: + await api_post(f"/instances/{instance_id}/halt") + return f"Stop command sent to instance {instance_id}." + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def reboot_instance(instance_id: str) -> str: + """Reboot a Vultr instance.""" + try: + await api_post(f"/instances/{instance_id}/reboot") + return f"Reboot command sent to instance {instance_id}." + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +# --------------------------------------------------------------------------- +# SSH Keys +# --------------------------------------------------------------------------- + +@mcp.tool() +async def list_ssh_keys() -> str: + """List all SSH keys on the account.""" + try: + data = await api_get("/ssh-keys") + keys = data.get("ssh_keys", []) + if not keys: + return "No SSH keys found." + lines = [f"{'ID':<40} {'Name':<30} {'Created'}"] + lines.append("-" * 80) + for k in keys: + lines.append(f"{k['id']:<40} {k['name']:<30} {k.get('date_created', 'n/a')}") + return f"SSH Keys ({len(keys)} total):\n" + "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def create_ssh_key(name: str, ssh_key: str) -> str: + """Add a new SSH key to the account.""" + try: + data = await api_post("/ssh-keys", {"name": name, "ssh_key": ssh_key}) + key = data.get("ssh_key", {}) + return ( + f"SSH key added.\n" + f"ID: {key['id']}\n" + f"Name: {key['name']}\n" + f"Created: {key.get('date_created', 'n/a')}\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +# --------------------------------------------------------------------------- +# Firewalls +# --------------------------------------------------------------------------- + +@mcp.tool() +async def list_firewalls() -> str: + """List all firewall groups.""" + try: + data = await api_get("/firewalls") + groups = data.get("firewall_groups", []) + if not groups: + return "No firewall groups found." + lines = [f"{'ID':<40} {'Description':<30} {'Rules':>6} {'Instances':>10}"] + lines.append("-" * 90) + for g in groups: + lines.append( + f"{g['id']:<40} {g.get('description', ''):<30} " + f"{g.get('rule_count', 0):>6} {g.get('instance_count', 0):>10}" + ) + return f"Firewall Groups ({len(groups)} total):\n" + "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def get_firewall(firewall_group_id: str) -> str: + """Get details and rules for a firewall group.""" + try: + group_data = await api_get(f"/firewalls/{firewall_group_id}") + rules_data = await api_get(f"/firewalls/{firewall_group_id}/rules") + g = group_data.get("firewall_group", {}) + rules = rules_data.get("firewall_rules", []) + lines = [ + f"Firewall Group: {g.get('description', '(no description)')}", + f"ID: {g['id']}", + f"Rules: {g.get('rule_count', 0)}", + f"Instances: {g.get('instance_count', 0)}", + "", + "Rules:", + f" {'#':>4} {'Action':<8} {'Proto':<8} {'Port':<10} {'Subnet':<20} {'Size':>6} {'Dir':<8}", + " " + "-" * 65, + ] + for r in rules: + lines.append( + f" {r['id']:>4} {r.get('action', 'accept'):<8} {r['protocol']:<8} " + f"{r.get('port', 'all'):<10} {r.get('subnet', 'any'):<20} " + f"{r.get('subnet_size', 0):>6} {r.get('direction', 'in'):<8}" + ) + if not rules: + lines.append(" (no rules)") + return "\n".join(lines) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def create_firewall(description: str) -> str: + """Create a new firewall group.""" + try: + data = await api_post("/firewalls", {"description": description}) + g = data.get("firewall_group", {}) + return ( + f"Firewall group created.\n" + f"ID: {g['id']}\n" + f"Description: {g.get('description')}\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +async def create_firewall_rule(firewall_group_id: str, protocol: str, port: str, subnet: str, subnet_size: int) -> str: + """Add a rule to a firewall group. Protocol: tcp/udp/icmp/gre/esp/ah. Direction is inbound.""" + try: + body = { + "ip_type": "v4", + "protocol": protocol, + "subnet": subnet, + "subnet_size": subnet_size, + "port": port, + } + data = await api_post(f"/firewalls/{firewall_group_id}/rules", body) + r = data.get("firewall_rule", {}) + return ( + f"Firewall rule added.\n" + f"Rule ID: {r['id']}\n" + f"Protocol: {r['protocol']}\n" + f"Port: {r.get('port', 'all')}\n" + f"Subnet: {r.get('subnet')}/{r.get('subnet_size')}\n" + f"Action: {r.get('action', 'accept')}\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +# --------------------------------------------------------------------------- +# Account +# --------------------------------------------------------------------------- + +@mcp.tool() +async def get_account() -> str: + """Get account information and balance.""" + try: + data = await api_get("/account") + acct = data.get("account", {}) + return ( + f"Account: {acct.get('name', 'n/a')}\n" + f"Email: {acct.get('email', 'n/a')}\n" + f"Balance: ${acct.get('balance', 0):.2f}\n" + f"Pending: ${acct.get('pending_charges', 0):.2f}\n" + f"ACLs: {', '.join(acct.get('acls', [])) or '(none)'}\n" + ) + except ValueError as e: + return f"Configuration error: {e}" + except httpx.HTTPStatusError as e: + return f"API error {e.response.status_code}: {e.response.text}" + except Exception as e: + return f"Error: {e}" + + +if __name__ == "__main__": + mcp.run(transport="stdio")