feat: initial Vultr MCP server — 16 tools for VPS management

2026-04-26 09:34:29 +10:00 · 2026-04-26 09:34:29 +10:00 · 325a4dffdf
commit 325a4dffdf
4 changed files with 542 additions and 0 deletions
--- a/.env.example
+++ b/.env.example
@ -0,0 +1 @@
+VULTR_API_KEY=your_api_key_here
--- a/README.md
+++ 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.
--- a/requirements.txt
+++ b/requirements.txt
@ -0,0 +1,2 @@
+mcp>=1.0.0
+httpx>=0.27.0
--- a/server.py
+++ 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")