452 lines
17 KiB
Python
452 lines
17 KiB
Python
import os
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("vultr", instructions="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")
|