diff --git a/src/vultr_dns_mcp/server.py b/src/vultr_dns_mcp/server.py index a1c8433..e470144 100644 --- a/src/vultr_dns_mcp/server.py +++ b/src/vultr_dns_mcp/server.py @@ -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)} - } - - @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.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" + ) + ] + + @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 } - } - - @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": { + return str(capabilities) + + 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. - - This tool retrieves all domains currently managed through Vultr DNS. - Each domain object includes domain name, creation date, and status information. - """ + @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"] + } + ) + ] + + @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)}] - - @mcp.tool() - async def get_dns_domain(domain: str) -> Dict[str, Any]: - """ - Get detailed information for a specific DNS domain. - - 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)} - - @mcp.tool() - async def create_dns_domain(domain: str, ip: str) -> Dict[str, Any]: - """ - Create a new DNS domain with a default A record. - - 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)} - - @mcp.tool() - async def delete_dns_domain(domain: str) -> Dict[str, Any]: - """ - Delete a DNS domain and ALL its associated records. - - ⚠️ WARNING: This permanently deletes the domain and all DNS records. - - 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)} - - @mcp.tool() - async def list_dns_records(domain: str) -> List[Dict[str, Any]]: - """ - List all DNS records for a specific domain. - - 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)}] - - @mcp.tool() - async def get_dns_record(domain: str, record_id: str) -> Dict[str, Any]: - """ - Get detailed information for a specific DNS record. - - 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)} - - @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. - - 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)} - - @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. - - 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)} - - @mcp.tool() - async def delete_dns_record(domain: str, record_id: str) -> Dict[str, Any]: - """ - Delete a specific DNS record. - - 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)} - - @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. - - 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": [] - } - - # 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)}") - - record_type = record_type.upper() - - # 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") - - # 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") - - elif record_type == 'AAAA': - if '::' in data and data.count('::') > 1: - validation_result["valid"] = False - validation_result["errors"].append("Invalid IPv6 address: multiple :: sequences") - - elif record_type == 'CNAME': - if name == '@' or name == '': - validation_result["valid"] = False - validation_result["errors"].append("CNAME records cannot be used for root domain (@)") - - 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") - - 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'") - - return { - "record_type": record_type, - "name": name, - "data": data, - "ttl": ttl, - "priority": priority, - "validation": validation_result - } - - @mcp.tool() - async def analyze_dns_records(domain: str) -> Dict[str, Any]: - """ - Analyze DNS configuration for a domain and provide insights. - - Args: - domain: The domain name to analyze (e.g., "example.com") - """ - try: - records = await vultr_client.list_records(domain) + if name == "list_dns_domains": + domains = await vultr_client.list_domains() + return [TextContent(type="text", text=str(domains))] - # Analyze records - record_types = {} - total_records = len(records) - ttl_values = [] - has_root_a = False - has_www = False - has_mx = False - has_spf = False + elif name == "get_dns_domain": + domain = arguments["domain"] + result = await vultr_client.get_domain(domain) + return [TextContent(type="text", text=str(result))] - 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) + 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))] + + elif name == "delete_dns_domain": + domain = arguments["domain"] + await vultr_client.delete_domain(domain) + return [TextContent(type="text", text=f"Domain {domain} deleted successfully")] + + elif name == "list_dns_records": + domain = arguments["domain"] + records = await vultr_client.list_records(domain) + return [TextContent(type="text", text=str(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))] + + 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))] + + 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))] + + 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")] + + elif name == "validate_dns_record": + record_type = arguments["record_type"] + name = arguments["name"] + data = arguments["data"] + ttl = arguments.get("ttl") + priority = arguments.get("priority") - record_types[record_type] = record_types.get(record_type, 0) + 1 - ttl_values.append(ttl) + validation_result = { + "valid": True, + "errors": [], + "warnings": [], + "suggestions": [] + } - 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 + # 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)}") + + record_type = record_type.upper() + + # 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") + + # 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") + + elif record_type == 'AAAA': + if '::' in data and data.count('::') > 1: + validation_result["valid"] = False + validation_result["errors"].append("Invalid IPv6 address: multiple :: sequences") + + elif record_type == 'CNAME': + if name == '@' or name == '': + validation_result["valid"] = False + validation_result["errors"].append("CNAME records cannot be used for root domain (@)") + + 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") + + 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'") + + result = { + "record_type": record_type, + "name": name, + "data": data, + "ttl": ttl, + "priority": priority, + "validation": validation_result + } + return [TextContent(type="text", text=str(result))] - # 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 - } + elif name == "analyze_dns_records": + domain = arguments["domain"] + 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") + + 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))] + else: + return [TextContent(type="text", text=f"Unknown tool: {name}")] + except Exception as e: - return {"error": str(e), "domain": domain} - - return mcp + return [TextContent(type="text", text=f"Error: {str(e)}")] + + 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)