Files
mcp-vultr/src/vultr_mcp/server.py

828 lines
28 KiB
Python

"""
MCP Server for Vultr Cloud Services
"""
import json
import logging
from typing import Dict, List, Any, Optional
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from .client import VultrClient
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Vultr MCP Server", description="MCP Server for Vultr Cloud Services")
# Initialize Vultr client
vultr_client = None
try:
vultr_client = VultrClient()
except ValueError as e:
logger.warning(f"Vultr client initialization failed: {e}")
logger.warning("Server will start but Vultr operations will fail until API key is configured")
# MCP Protocol Models
class Tool(BaseModel):
"""MCP Tool definition"""
name: str
description: str
inputSchema: Dict[str, Any]
class Resource(BaseModel):
"""MCP Resource definition"""
uri: str
name: str
description: str
mimeType: str = "text/plain"
class Prompt(BaseModel):
"""MCP Prompt definition"""
name: str
description: str
arguments: List[Dict[str, Any]] = []
# Request/Response Models
class ToolCallRequest(BaseModel):
"""Request for tool execution"""
name: str
arguments: Dict[str, Any] = {}
class ToolCallResponse(BaseModel):
"""Response from tool execution"""
content: List[Dict[str, Any]]
isError: bool = False
# MCP Protocol Endpoints
@app.get("/")
async def root():
"""Root endpoint"""
return {
"name": "Vultr MCP Server",
"version": "1.0.0",
"description": "Model Context Protocol Server for Vultr Cloud Services"
}
@app.get("/tools")
async def list_tools() -> List[Tool]:
"""List available tools"""
tools = [
Tool(
name="get_account_info",
description="Get Vultr account information",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="list_instances",
description="List all Vultr instances",
inputSchema={
"type": "object",
"properties": {
"per_page": {
"type": "integer",
"description": "Number of instances per page",
"default": 100
}
}
}
),
Tool(
name="get_instance",
description="Get details of a specific instance",
inputSchema={
"type": "object",
"properties": {
"instance_id": {
"type": "string",
"description": "ID of the instance"
}
},
"required": ["instance_id"]
}
),
Tool(
name="create_instance",
description="Create a new Vultr instance",
inputSchema={
"type": "object",
"properties": {
"region": {
"type": "string",
"description": "Region code (e.g., 'ams')"
},
"plan": {
"type": "string",
"description": "Plan ID (e.g., 'vc2-1c-1gb')"
},
"os_id": {
"type": "integer",
"description": "Operating System ID"
},
"label": {
"type": "string",
"description": "Instance label",
"default": ""
},
"hostname": {
"type": "string",
"description": "Instance hostname",
"default": ""
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Tags for the instance",
"default": []
}
},
"required": ["region", "plan", "os_id"]
}
),
Tool(
name="delete_instance",
description="Delete a Vultr instance",
inputSchema={
"type": "object",
"properties": {
"instance_id": {
"type": "string",
"description": "ID of the instance to delete"
}
},
"required": ["instance_id"]
}
),
Tool(
name="start_instance",
description="Start a Vultr instance",
inputSchema={
"type": "object",
"properties": {
"instance_id": {
"type": "string",
"description": "ID of the instance to start"
}
},
"required": ["instance_id"]
}
),
Tool(
name="stop_instance",
description="Stop a Vultr instance",
inputSchema={
"type": "object",
"properties": {
"instance_id": {
"type": "string",
"description": "ID of the instance to stop"
}
},
"required": ["instance_id"]
}
),
Tool(
name="reboot_instance",
description="Reboot a Vultr instance",
inputSchema={
"type": "object",
"properties": {
"instance_id": {
"type": "string",
"description": "ID of the instance to reboot"
}
},
"required": ["instance_id"]
}
),
Tool(
name="list_regions",
description="List available Vultr regions",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="list_plans",
description="List available Vultr plans",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="list_os",
description="List available Vultr operating systems",
inputSchema={"type": "object", "properties": {}}
),
),
# DNS Management Tools
Tool(
name="list_dns_domains",
description="List all DNS domains",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="create_dns_domain",
description="Create a new DNS domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name (e.g., 'example.com')"
},
"ip": {
"type": "string",
"description": "IP address for the domain"
}
},
"required": ["domain", "ip"]
}
),
Tool(
name="delete_dns_domain",
description="Delete a DNS domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="get_dns_domain",
description="Get DNS domain details",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="list_dns_records",
description="List DNS records for a domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="create_dns_record",
description="Create a new DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"type": {
"type": "string",
"description": "Record type (A, AAAA, CNAME, MX, TXT, etc.)"
},
"name": {
"type": "string",
"description": "Record name (e.g., 'www', '@', 'mail')"
},
"data": {
"type": "string",
"description": "Record data (IP address, hostname, etc.)"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds",
"default": 300
},
"priority": {
"type": "integer",
"description": "Priority for MX records",
"default": 0
}
},
"required": ["domain", "type", "name", "data"]
}
),
Tool(
name="update_dns_record",
description="Update a DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"record_id": {
"type": "string",
"description": "Record ID"
},
"type": {
"type": "string",
"description": "Record type (A, AAAA, CNAME, MX, TXT, etc.)"
},
"name": {
"type": "string",
"description": "Record name"
},
"data": {
"type": "string",
"description": "Record data"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds"
},
"priority": {
"type": "integer",
"description": "Priority for MX records"
}
},
"required": ["domain", "record_id"]
}
),
Tool(
name="delete_dns_record",
description="Delete a DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"record_id": {
"type": "string",
"description": "Record ID"
}
},
"required": ["domain", "record_id"]
}
)
]
return tools
@app.get("/resources")
async def list_resources() -> List[Resource]:
"""List available resources"""
resources = [
Resource(
uri="vultr://account/info",
name="Vultr Account Info",
description="Vultr account information"
),
Resource(
uri="vultr://instances/list",
name="Vultr Instances",
description="List of Vultr instances"
),
Resource(
uri="vultr://regions/list",
name="Vultr Regions",
description="List of available Vultr regions"
),
Resource(
uri="vultr://plans/list",
name="Vultr Plans",
description="List of available Vultr plans"
),
Resource(
uri="vultr://os/list",
name="Vultr Operating Systems",
description="List of available Vultr operating systems"
),
Resource(
uri="vultr://dns/domains",
name="Vultr DNS Domains",
description="List of DNS domains"
),
Resource(
uri="vultr://dns/records",
name="Vultr DNS Records",
description="DNS records for a domain"
),
]
return resources
@app.get("/prompts")
async def list_prompts() -> List[Prompt]:
"""List available prompts"""
prompts = [
Prompt(
name="deploy_web_server",
description="Deploy a web server on Vultr",
arguments=[
{"name": "region", "description": "Region to deploy in", "required": True},
{"name": "plan", "description": "Server plan", "required": True},
{"name": "domain", "description": "Domain name", "required": False},
]
),
Prompt(
name="manage_instances",
description="Manage Vultr instances",
arguments=[
{"name": "action", "description": "Action to perform (start, stop, reboot, delete)", "required": True},
{"name": "instance_id", "description": "Instance ID", "required": True},
]
),
Prompt(
name="monitor_resources",
description="Monitor Vultr resources",
arguments=[
{"name": "resource_type", "description": "Type of resource to monitor", "required": False},
]
),
]
return prompts
@app.post("/tools/call")
async def call_tool(request: ToolCallRequest) -> ToolCallResponse:
"""Execute a tool"""
if not vultr_client:
return ToolCallResponse(
content=[{"type": "text", "text": "Vultr client not initialized. Please set VULTR_API_KEY environment variable."}],
isError=True
)
try:
result = None
if request.name == "get_account_info":
result = vultr_client.get_account_info()
elif request.name == "list_instances":
per_page = request.arguments.get("per_page", 100)
result = vultr_client.list_instances(per_page=per_page)
elif request.name == "get_instance":
instance_id = request.arguments.get("instance_id")
if not instance_id:
raise ValueError("instance_id is required")
result = vultr_client.get_instance(instance_id)
elif request.name == "create_instance":
result = vultr_client.create_instance(**request.arguments)
elif request.name == "delete_instance":
instance_id = request.arguments.get("instance_id")
if not instance_id:
raise ValueError("instance_id is required")
result = vultr_client.delete_instance(instance_id)
elif request.name == "start_instance":
instance_id = request.arguments.get("instance_id")
if not instance_id:
raise ValueError("instance_id is required")
result = vultr_client.start_instance(instance_id)
elif request.name == "stop_instance":
instance_id = request.arguments.get("instance_id")
if not instance_id:
raise ValueError("instance_id is required")
result = vultr_client.stop_instance(instance_id)
elif request.name == "reboot_instance":
instance_id = request.arguments.get("instance_id")
if not instance_id:
raise ValueError("instance_id is required")
result = vultr_client.reboot_instance(instance_id)
elif request.name == "list_regions":
result = vultr_client.list_regions()
elif request.name == "list_plans":
result = vultr_client.list_plans()
elif request.name == "list_os":
result = vultr_client.list_os()
elif request.name == "list_dns_domains":
result = vultr_client.list_dns_domains()
elif request.name == "create_dns_domain":
domain = request.arguments.get("domain")
ip = request.arguments.get("ip")
if not domain or not ip:
raise ValueError("domain and ip are required")
result = vultr_client.create_dns_domain(domain, ip)
elif request.name == "delete_dns_domain":
domain = request.arguments.get("domain")
if not domain:
raise ValueError("domain is required")
result = vultr_client.delete_dns_domain(domain)
elif request.name == "get_dns_domain":
domain = request.arguments.get("domain")
if not domain:
raise ValueError("domain is required")
result = vultr_client.get_dns_domain(domain)
elif request.name == "list_dns_records":
domain = request.arguments.get("domain")
if not domain:
raise ValueError("domain is required")
result = vultr_client.list_dns_records(domain)
elif request.name == "create_dns_record":
domain = request.arguments.get("domain")
record_type = request.arguments.get("type")
name = request.arguments.get("name")
data = request.arguments.get("data")
ttl = request.arguments.get("ttl", 300)
priority = request.arguments.get("priority", 0)
if not domain or not record_type or not name or not data:
raise ValueError("domain, type, name, and data are required")
result = vultr_client.create_dns_record(domain, record_type, name, data, ttl, priority)
elif request.name == "update_dns_record":
domain = request.arguments.get("domain")
record_id = request.arguments.get("record_id")
if not domain or not record_id:
raise ValueError("domain and record_id are required")
# Filter out None values
update_args = {k: v for k, v in request.arguments.items()
if k not in ["domain", "record_id"] and v is not None}
result = vultr_client.update_dns_record(domain, record_id, **update_args)
elif request.name == "delete_dns_record":
domain = request.arguments.get("domain")
record_id = request.arguments.get("record_id")
if not domain or not record_id:
raise ValueError("domain and record_id are required")
result = vultr_client.delete_dns_record(domain, record_id)
else:
return ToolCallResponse(
content=[{"type": "text", "text": f"Unknown tool: {request.name}"}],
isError=True
)
# Format the result
if isinstance(result, dict) and 'error' in result:
return ToolCallResponse(
content=[{"type": "text", "text": f"Error: {result['error']}"}],
isError=True
)
# Convert result to JSON string for display
result_str = json.dumps(result, indent=2, default=str)
return ToolCallResponse(
content=[{"type": "text", "text": result_str}]
)
Prompt(
name="manage_dns",
description="Manage DNS domains and records",
arguments=[
{"name": "action", "description": "Action to perform (create_domain, delete_domain, list_records, create_record, update_record, delete_record)", "required": True},
{"name": "domain", "description": "Domain name", "required": True},
{"name": "record_id", "description": "Record ID (for update/delete)", "required": False},
{"name": "record_type", "description": "Record type (A, AAAA, CNAME, etc.)", "required": False},
{"name": "name", "description": "Record name", "required": False},
{"name": "data", "description": "Record data", "required": False},
{"name": "ip", "description": "IP address (for create domain)", "required": False},
]
),
except Exception as e:
logger.error(f"Error executing tool {request.name}: {e}", exc_info=True)
return ToolCallResponse(
content=[{"type": "text", "text": f"Error: {str(e)}"}],
isError=True
)
@app.get("/resources/{resource_uri:path}")
async def get_resource(resource_uri: str):
"""Get resource content"""
if not vultr_client:
return JSONResponse(
status_code=500,
content={"error": "Vultr client not initialized"}
)
try:
result = None
if resource_uri == "vultr://account/info":
result = vultr_client.get_account_info()
elif resource_uri == "vultr://instances/list":
result = vultr_client.list_instances()
elif resource_uri == "vultr://regions/list":
result = vultr_client.list_regions()
elif resource_uri == "vultr://plans/list":
result = vultr_client.list_plans()
elif resource_uri == "vultr://os/list":
result = vultr_client.list_os()
elif resource_uri == "vultr://dns/domains":
result = vultr_client.list_dns_domains()
elif resource_uri == "vultr://dns/records":
# This requires a domain parameter, so we'll return instructions
result = {"message": "Please use the list_dns_records tool with domain parameter to get DNS records"}
else:
return JSONResponse(
status_code=404,
content={"error": f"Resource not found: {resource_uri}"}
)
if isinstance(result, dict) and 'error' in result:
return JSONResponse(
status_code=500,
content={"error": result['error']}
)
return JSONResponse(content=result)
except Exception as e:
logger.error(f"Error getting resource {resource_uri}: {e}", exc_info=True)
return JSONResponse(
status_code=500,
content={"error": str(e)}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
# DNS Management Tools
Tool(
name="list_dns_domains",
description="List all DNS domains",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="create_dns_domain",
description="Create a new DNS domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name (e.g., 'example.com')"
},
"ip": {
"type": "string",
"description": "IP address for the domain"
}
},
"required": ["domain", "ip"]
}
),
Tool(
name="delete_dns_domain",
description="Delete a DNS domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="get_dns_domain",
description="Get DNS domain details",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="list_dns_records",
description="List DNS records for a domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
}
},
"required": ["domain"]
}
),
Tool(
name="create_dns_record",
description="Create a new DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"type": {
"type": "string",
"description": "Record type (A, AAAA, CNAME, MX, TXT, etc.)"
},
"name": {
"type": "string",
"description": "Record name (e.g., 'www', '@', 'mail')"
},
"data": {
"type": "string",
"description": "Record data (IP address, hostname, etc.)"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds",
"default": 300
},
"priority": {
"type": "integer",
"description": "Priority for MX records",
"default": 0
}
},
"required": ["domain", "type", "name", "data"]
}
),
Tool(
name="update_dns_record",
description="Update a DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"record_id": {
"type": "string",
"description": "Record ID"
},
"type": {
"type": "string",
"description": "Record type (A, AAAA, CNAME, MX, TXT, etc.)"
},
"name": {
"type": "string",
"description": "Record name"
},
"data": {
"type": "string",
"description": "Record data"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds"
},
"priority": {
"type": "integer",
"description": "Priority for MX records"
}
},
"required": ["domain", "record_id"]
}
),
Tool(
name="delete_dns_record",
description="Delete a DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Domain name"
},
"record_id": {
"type": "string",
"description": "Record ID"
}
},
"required": ["domain", "record_id"]
}
),