MCP/mcp-vultr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Migrate from fastmcp to official MCP package

Browse Source
This commit is contained in:
Ryan Malloy 2025-06-11 17:53:43 -06:00
parent 5c87097158
commit b8fd6e4632
1 changed files with 485 additions and 369 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

794
src/vultr_dns_mcp/server.py
View File

@ -10,7 +10,9 @@ import re
from typing import Any, Dict, List, Optional
import httpx
from fastmcp import FastMCP
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent
from pydantic import BaseModel
@ -143,15 +145,15 @@ class VultrDNSServer:
return await self._make_request("DELETE", f"/domains/{domain}/records/{record_id}")
def create_mcp_server(api_key: Optional[str] = None) -> FastMCP:
def create_mcp_server(api_key: Optional[str] = None) -> Server:
"""
Create and configure a FastMCP server for Vultr DNS management.
Create and configure an MCP server for Vultr DNS management.
Args:
api_key: Vultr API key. If not provided, will read from VULTR_API_KEY env var.
Returns:
Configured FastMCP server instance
Configured MCP server instance
Raises:
ValueError: If API key is not provided and not found in environment
@ -164,43 +166,43 @@ def create_mcp_server(api_key: Optional[str] = None) -> FastMCP:
"VULTR_API_KEY must be provided either as parameter or environment variable"
)
# Initialize FastMCP server
mcp = FastMCP("Vultr DNS Manager")
# Initialize MCP server
server = Server("vultr-dns-mcp")
# Initialize Vultr client
vultr_client = VultrDNSServer(api_key)
# Add resources for client discovery
@mcp.resource("vultr://domains")
async def get_domains_resource():
"""Resource listing all available domains."""
try:
domains = await vultr_client.list_domains()
return {
"uri": "vultr://domains",
"name": "DNS Domains",
"description": "All DNS domains in your Vultr account",
"mimeType": "application/json",
"content": domains
}
except Exception as e:
return {
"uri": "vultr://domains",
"name": "DNS Domains",
"description": f"Error loading domains: {str(e)}",
"mimeType": "application/json",
"content": {"error": str(e)}
}
@server.list_resources()
async def list_resources() -> List[Resource]:
"""List available resources."""
return [
Resource(
uri="vultr://domains",
name="DNS Domains",
description="All DNS domains in your Vultr account",
mimeType="application/json"
),
Resource(
uri="vultr://capabilities",
name="Server Capabilities",
description="Vultr DNS server capabilities and supported features",
mimeType="application/json"
)
]
@mcp.resource("vultr://capabilities")
async def get_capabilities_resource():
"""Resource describing server capabilities and supported record types."""
return {
"uri": "vultr://capabilities",
"name": "Server Capabilities",
"description": "Vultr DNS server capabilities and supported features",
"mimeType": "application/json",
"content": {
@server.read_resource()
async def read_resource(uri: str) -> str:
"""Read a specific resource."""
if uri == "vultr://domains":
try:
domains = await vultr_client.list_domains()
return str(domains)
except Exception as e:
return f"Error loading domains: {str(e)}"
elif uri == "vultr://capabilities":
capabilities = {
"supported_record_types": [
{
"type": "A",
@ -253,356 +255,470 @@ def create_mcp_server(api_key: Optional[str] = None) -> FastMCP:
"min_ttl": 60,
"max_ttl": 86400
}
}
return str(capabilities)
@mcp.resource("vultr://records/{domain}")
async def get_domain_records_resource(domain: str):
"""Resource listing all records for a specific domain."""
try:
records = await vultr_client.list_records(domain)
return {
"uri": f"vultr://records/{domain}",
"name": f"DNS Records for {domain}",
"description": f"All DNS records configured for domain {domain}",
"mimeType": "application/json",
"content": {
elif uri.startswith("vultr://records/"):
domain = uri.replace("vultr://records/", "")
try:
records = await vultr_client.list_records(domain)
return str({
"domain": domain,
"records": records,
"record_count": len(records)
}
}
except Exception as e:
return {
"uri": f"vultr://records/{domain}",
"name": f"DNS Records for {domain}",
"description": f"Error loading records for {domain}: {str(e)}",
"mimeType": "application/json",
"content": {"error": str(e), "domain": domain}
}
})
except Exception as e:
return f"Error loading records for {domain}: {str(e)}"
return "Resource not found"
# Define MCP tools
@mcp.tool()
async def list_dns_domains() -> List[Dict[str, Any]]:
"""
List all DNS domains in your Vultr account.
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List available tools."""
return [
Tool(
name="list_dns_domains",
description="List all DNS domains in your Vultr account",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_dns_domain",
description="Get detailed information for a specific DNS domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name to retrieve (e.g., 'example.com')"
}
},
"required": ["domain"]
}
),
Tool(
name="create_dns_domain",
description="Create a new DNS domain with a default A record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name to create (e.g., 'newdomain.com')"
},
"ip": {
"type": "string",
"description": "IPv4 address for the default A record (e.g., '192.168.1.100')"
}
},
"required": ["domain", "ip"]
}
),
Tool(
name="delete_dns_domain",
description="Delete a DNS domain and ALL its associated records",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name to delete (e.g., 'example.com')"
}
},
"required": ["domain"]
}
),
Tool(
name="list_dns_records",
description="List all DNS records for a specific domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name (e.g., 'example.com')"
}
},
"required": ["domain"]
}
),
Tool(
name="get_dns_record",
description="Get detailed information for a specific DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name (e.g., 'example.com')"
},
"record_id": {
"type": "string",
"description": "The unique record identifier"
}
},
"required": ["domain", "record_id"]
}
),
Tool(
name="create_dns_record",
description="Create a new DNS record for a domain",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name (e.g., 'example.com')"
},
"record_type": {
"type": "string",
"description": "Record type (A, AAAA, CNAME, MX, TXT, NS, SRV)"
},
"name": {
"type": "string",
"description": "Record name/subdomain"
},
"data": {
"type": "string",
"description": "Record value"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds (60-86400, default: 300)"
},
"priority": {
"type": "integer",
"description": "Priority for MX/SRV records (0-65535)"
}
},
"required": ["domain", "record_type", "name", "data"]
}
),
Tool(
name="update_dns_record",
description="Update an existing DNS record with new configuration",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name (e.g., 'example.com')"
},
"record_id": {
"type": "string",
"description": "The unique identifier of the record to update"
},
"record_type": {
"type": "string",
"description": "New record type (A, AAAA, CNAME, MX, TXT, NS, SRV)"
},
"name": {
"type": "string",
"description": "New record name/subdomain"
},
"data": {
"type": "string",
"description": "New record value"
},
"ttl": {
"type": "integer",
"description": "New TTL in seconds (60-86400, optional)"
},
"priority": {
"type": "integer",
"description": "New priority for MX/SRV records (optional)"
}
},
"required": ["domain", "record_id", "record_type", "name", "data"]
}
),
Tool(
name="delete_dns_record",
description="Delete a specific DNS record",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name (e.g., 'example.com')"
},
"record_id": {
"type": "string",
"description": "The unique identifier of the record to delete"
}
},
"required": ["domain", "record_id"]
}
),
Tool(
name="validate_dns_record",
description="Validate DNS record parameters before creation",
inputSchema={
"type": "object",
"properties": {
"record_type": {
"type": "string",
"description": "The record type (A, AAAA, CNAME, MX, TXT, NS, SRV)"
},
"name": {
"type": "string",
"description": "The record name/subdomain"
},
"data": {
"type": "string",
"description": "The record data/value"
},
"ttl": {
"type": "integer",
"description": "Time to live in seconds (optional)"
},
"priority": {
"type": "integer",
"description": "Priority for MX/SRV records (optional)"
}
},
"required": ["record_type", "name", "data"]
}
),
Tool(
name="analyze_dns_records",
description="Analyze DNS configuration for a domain and provide insights",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name to analyze (e.g., 'example.com')"
}
},
"required": ["domain"]
}
)
]
This tool retrieves all domains currently managed through Vultr DNS.
Each domain object includes domain name, creation date, and status information.
"""
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls."""
try:
domains = await vultr_client.list_domains()
return domains
except Exception as e:
return [{"error": str(e)}]
if name == "list_dns_domains":
domains = await vultr_client.list_domains()
return [TextContent(type="text", text=str(domains))]
@mcp.tool()
async def get_dns_domain(domain: str) -> Dict[str, Any]:
"""
Get detailed information for a specific DNS domain.
elif name == "get_dns_domain":
domain = arguments["domain"]
result = await vultr_client.get_domain(domain)
return [TextContent(type="text", text=str(result))]
Args:
domain: The domain name to retrieve (e.g., "example.com")
"""
try:
return await vultr_client.get_domain(domain)
except Exception as e:
return {"error": str(e)}
elif name == "create_dns_domain":
domain = arguments["domain"]
ip = arguments["ip"]
result = await vultr_client.create_domain(domain, ip)
return [TextContent(type="text", text=str(result))]
@mcp.tool()
async def create_dns_domain(domain: str, ip: str) -> Dict[str, Any]:
"""
Create a new DNS domain with a default A record.
elif name == "delete_dns_domain":
domain = arguments["domain"]
await vultr_client.delete_domain(domain)
return [TextContent(type="text", text=f"Domain {domain} deleted successfully")]
Args:
domain: The domain name to create (e.g., "newdomain.com")
ip: IPv4 address for the default A record (e.g., "192.168.1.100")
"""
try:
return await vultr_client.create_domain(domain, ip)
except Exception as e:
return {"error": str(e)}
elif name == "list_dns_records":
domain = arguments["domain"]
records = await vultr_client.list_records(domain)
return [TextContent(type="text", text=str(records))]
@mcp.tool()
async def delete_dns_domain(domain: str) -> Dict[str, Any]:
"""
Delete a DNS domain and ALL its associated records.
elif name == "get_dns_record":
domain = arguments["domain"]
record_id = arguments["record_id"]
result = await vultr_client.get_record(domain, record_id)
return [TextContent(type="text", text=str(result))]
⚠️ WARNING: This permanently deletes the domain and all DNS records.
elif name == "create_dns_record":
domain = arguments["domain"]
record_type = arguments["record_type"]
name = arguments["name"]
data = arguments["data"]
ttl = arguments.get("ttl")
priority = arguments.get("priority")
result = await vultr_client.create_record(domain, record_type, name, data, ttl, priority)
return [TextContent(type="text", text=str(result))]
Args:
domain: The domain name to delete (e.g., "example.com")
"""
try:
await vultr_client.delete_domain(domain)
return {"success": f"Domain {domain} deleted successfully"}
except Exception as e:
return {"error": str(e)}
elif name == "update_dns_record":
domain = arguments["domain"]
record_id = arguments["record_id"]
record_type = arguments["record_type"]
name = arguments["name"]
data = arguments["data"]
ttl = arguments.get("ttl")
priority = arguments.get("priority")
result = await vultr_client.update_record(domain, record_id, record_type, name, data, ttl, priority)
return [TextContent(type="text", text=str(result))]
@mcp.tool()
async def list_dns_records(domain: str) -> List[Dict[str, Any]]:
"""
List all DNS records for a specific domain.
elif name == "delete_dns_record":
domain = arguments["domain"]
record_id = arguments["record_id"]
await vultr_client.delete_record(domain, record_id)
return [TextContent(type="text", text=f"DNS record {record_id} deleted successfully")]
Args:
domain: The domain name (e.g., "example.com")
"""
try:
records = await vultr_client.list_records(domain)
return records
except Exception as e:
return [{"error": str(e)}]
elif name == "validate_dns_record":
record_type = arguments["record_type"]
name = arguments["name"]
data = arguments["data"]
ttl = arguments.get("ttl")
priority = arguments.get("priority")
@mcp.tool()
async def get_dns_record(domain: str, record_id: str) -> Dict[str, Any]:
"""
Get detailed information for a specific DNS record.
validation_result = {
"valid": True,
"errors": [],
"warnings": [],
"suggestions": []
}
Args:
domain: The domain name (e.g., "example.com")
record_id: The unique record identifier
"""
try:
return await vultr_client.get_record(domain, record_id)
except Exception as e:
return {"error": str(e)}
# Validate record type
valid_types = ['A', 'AAAA', 'CNAME', 'MX', 'TXT', 'NS', 'SRV']
if record_type.upper() not in valid_types:
validation_result["valid"] = False
validation_result["errors"].append(f"Invalid record type. Must be one of: {', '.join(valid_types)}")
@mcp.tool()
async def create_dns_record(
domain: str,
record_type: str,
name: str,
data: str,
ttl: Optional[int] = None,
priority: Optional[int] = None
) -> Dict[str, Any]:
"""
Create a new DNS record for a domain.
record_type = record_type.upper()
Args:
domain: The domain name (e.g., "example.com")
record_type: Record type (A, AAAA, CNAME, MX, TXT, NS, SRV)
name: Record name/subdomain
data: Record value
ttl: Time to live in seconds (60-86400, default: 300)
priority: Priority for MX/SRV records (0-65535)
"""
try:
return await vultr_client.create_record(domain, record_type, name, data, ttl, priority)
except Exception as e:
return {"error": str(e)}
# Validate TTL
if ttl is not None:
if ttl < 60 or ttl > 86400:
validation_result["warnings"].append("TTL should be between 60 and 86400 seconds")
elif ttl < 300:
validation_result["warnings"].append("Low TTL values may impact DNS performance")
@mcp.tool()
async def update_dns_record(
domain: str,
record_id: str,
record_type: str,
name: str,
data: str,
ttl: Optional[int] = None,
priority: Optional[int] = None
) -> Dict[str, Any]:
"""
Update an existing DNS record with new configuration.
# Record-specific validation
if record_type == 'A':
ipv4_pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
if not re.match(ipv4_pattern, data):
validation_result["valid"] = False
validation_result["errors"].append("Invalid IPv4 address format")
Args:
domain: The domain name (e.g., "example.com")
record_id: The unique identifier of the record to update
record_type: New record type (A, AAAA, CNAME, MX, TXT, NS, SRV)
name: New record name/subdomain
data: New record value
ttl: New TTL in seconds (60-86400, optional)
priority: New priority for MX/SRV records (optional)
"""
try:
return await vultr_client.update_record(domain, record_id, record_type, name, data, ttl, priority)
except Exception as e:
return {"error": str(e)}
elif record_type == 'AAAA':
if '::' in data and data.count('::') > 1:
validation_result["valid"] = False
validation_result["errors"].append("Invalid IPv6 address: multiple :: sequences")
@mcp.tool()
async def delete_dns_record(domain: str, record_id: str) -> Dict[str, Any]:
"""
Delete a specific DNS record.
elif record_type == 'CNAME':
if name == '@' or name == '':
validation_result["valid"] = False
validation_result["errors"].append("CNAME records cannot be used for root domain (@)")
Args:
domain: The domain name (e.g., "example.com")
record_id: The unique identifier of the record to delete
"""
try:
await vultr_client.delete_record(domain, record_id)
return {"success": f"DNS record {record_id} deleted successfully"}
except Exception as e:
return {"error": str(e)}
elif record_type == 'MX':
if priority is None:
validation_result["valid"] = False
validation_result["errors"].append("MX records require a priority value")
elif priority < 0 or priority > 65535:
validation_result["valid"] = False
validation_result["errors"].append("MX priority must be between 0 and 65535")
@mcp.tool()
async def validate_dns_record(
record_type: str,
name: str,
data: str,
ttl: Optional[int] = None,
priority: Optional[int] = None
) -> Dict[str, Any]:
"""
Validate DNS record parameters before creation.
elif record_type == 'SRV':
if priority is None:
validation_result["valid"] = False
validation_result["errors"].append("SRV records require a priority value")
srv_parts = data.split()
if len(srv_parts) != 3:
validation_result["valid"] = False
validation_result["errors"].append("SRV data must be in format: 'weight port target'")
Args:
record_type: The record type (A, AAAA, CNAME, MX, TXT, NS, SRV)
name: The record name/subdomain
data: The record data/value
ttl: Time to live in seconds (optional)
priority: Priority for MX/SRV records (optional)
"""
validation_result = {
"valid": True,
"errors": [],
"warnings": [],
"suggestions": []
}
result = {
"record_type": record_type,
"name": name,
"data": data,
"ttl": ttl,
"priority": priority,
"validation": validation_result
}
return [TextContent(type="text", text=str(result))]
# Validate record type
valid_types = ['A', 'AAAA', 'CNAME', 'MX', 'TXT', 'NS', 'SRV']
if record_type.upper() not in valid_types:
validation_result["valid"] = False
validation_result["errors"].append(f"Invalid record type. Must be one of: {', '.join(valid_types)}")
elif name == "analyze_dns_records":
domain = arguments["domain"]
records = await vultr_client.list_records(domain)
record_type = record_type.upper()
# Analyze records
record_types = {}
total_records = len(records)
ttl_values = []
has_root_a = False
has_www = False
has_mx = False
has_spf = False
# Validate TTL
if ttl is not None:
if ttl < 60 or ttl > 86400:
validation_result["warnings"].append("TTL should be between 60 and 86400 seconds")
elif ttl < 300:
validation_result["warnings"].append("Low TTL values may impact DNS performance")
for record in records:
record_type = record.get('type', 'UNKNOWN')
record_name = record.get('name', '')
record_data = record.get('data', '')
ttl = record.get('ttl', 300)
# Record-specific validation
if record_type == 'A':
ipv4_pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
if not re.match(ipv4_pattern, data):
validation_result["valid"] = False
validation_result["errors"].append("Invalid IPv4 address format")
record_types[record_type] = record_types.get(record_type, 0) + 1
ttl_values.append(ttl)
elif record_type == 'AAAA':
if '::' in data and data.count('::') > 1:
validation_result["valid"] = False
validation_result["errors"].append("Invalid IPv6 address: multiple :: sequences")
if record_type == 'A' and record_name in ['@', domain]:
has_root_a = True
if record_name == 'www':
has_www = True
if record_type == 'MX':
has_mx = True
if record_type == 'TXT' and 'spf1' in record_data.lower():
has_spf = True
elif record_type == 'CNAME':
if name == '@' or name == '':
validation_result["valid"] = False
validation_result["errors"].append("CNAME records cannot be used for root domain (@)")
# Generate recommendations
recommendations = []
issues = []
elif record_type == 'MX':
if priority is None:
validation_result["valid"] = False
validation_result["errors"].append("MX records require a priority value")
elif priority < 0 or priority > 65535:
validation_result["valid"] = False
validation_result["errors"].append("MX priority must be between 0 and 65535")
if not has_root_a:
recommendations.append("Consider adding an A record for the root domain (@)")
if not has_www:
recommendations.append("Consider adding a www subdomain (A or CNAME record)")
if not has_mx and total_records > 1:
recommendations.append("Consider adding MX records if you plan to use email")
if has_mx and not has_spf:
recommendations.append("Add SPF record (TXT) to prevent email spoofing")
elif record_type == 'SRV':
if priority is None:
validation_result["valid"] = False
validation_result["errors"].append("SRV records require a priority value")
srv_parts = data.split()
if len(srv_parts) != 3:
validation_result["valid"] = False
validation_result["errors"].append("SRV data must be in format: 'weight port target'")
avg_ttl = sum(ttl_values) / len(ttl_values) if ttl_values else 0
low_ttl_count = sum(1 for ttl in ttl_values if ttl < 300)
return {
"record_type": record_type,
"name": name,
"data": data,
"ttl": ttl,
"priority": priority,
"validation": validation_result
}
if low_ttl_count > total_records * 0.5:
issues.append("Many records have very low TTL values, which may impact performance")
@mcp.tool()
async def analyze_dns_records(domain: str) -> Dict[str, Any]:
"""
Analyze DNS configuration for a domain and provide insights.
result = {
"domain": domain,
"analysis": {
"total_records": total_records,
"record_types": record_types,
"average_ttl": round(avg_ttl),
"configuration_status": {
"has_root_domain": has_root_a,
"has_www_subdomain": has_www,
"has_email_mx": has_mx,
"has_spf_protection": has_spf
}
},
"recommendations": recommendations,
"potential_issues": issues,
"records_detail": records
}
return [TextContent(type="text", text=str(result))]
Args:
domain: The domain name to analyze (e.g., "example.com")
"""
try:
records = await vultr_client.list_records(domain)
# Analyze records
record_types = {}
total_records = len(records)
ttl_values = []
has_root_a = False
has_www = False
has_mx = False
has_spf = False
for record in records:
record_type = record.get('type', 'UNKNOWN')
record_name = record.get('name', '')
record_data = record.get('data', '')
ttl = record.get('ttl', 300)
record_types[record_type] = record_types.get(record_type, 0) + 1
ttl_values.append(ttl)
if record_type == 'A' and record_name in ['@', domain]:
has_root_a = True
if record_name == 'www':
has_www = True
if record_type == 'MX':
has_mx = True
if record_type == 'TXT' and 'spf1' in record_data.lower():
has_spf = True
# Generate recommendations
recommendations = []
issues = []
if not has_root_a:
recommendations.append("Consider adding an A record for the root domain (@)")
if not has_www:
recommendations.append("Consider adding a www subdomain (A or CNAME record)")
if not has_mx and total_records > 1:
recommendations.append("Consider adding MX records if you plan to use email")
if has_mx and not has_spf:
recommendations.append("Add SPF record (TXT) to prevent email spoofing")
avg_ttl = sum(ttl_values) / len(ttl_values) if ttl_values else 0
low_ttl_count = sum(1 for ttl in ttl_values if ttl < 300)
if low_ttl_count > total_records * 0.5:
issues.append("Many records have very low TTL values, which may impact performance")
return {
"domain": domain,
"analysis": {
"total_records": total_records,
"record_types": record_types,
"average_ttl": round(avg_ttl),
"configuration_status": {
"has_root_domain": has_root_a,
"has_www_subdomain": has_www,
"has_email_mx": has_mx,
"has_spf_protection": has_spf
}
},
"recommendations": recommendations,
"potential_issues": issues,
"records_detail": records
}
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except Exception as e:
return {"error": str(e), "domain": domain}
return [TextContent(type="text", text=f"Error: {str(e)}")]
return mcp
return server
def run_server(api_key: Optional[str] = None) -> None:
async def run_server(api_key: Optional[str] = None) -> None:
"""
Create and run a Vultr DNS MCP server.
Args:
api_key: Vultr API key. If not provided, will read from VULTR_API_KEY env var.
"""
mcp = create_mcp_server(api_key)
mcp.run()
server = create_mcp_server(api_key)
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, None)