DNSDumpster MCP Server

Instructions

Implementation Reference

• server.py:190-228 (handler)

  The core handler function for the 'query_domain' tool. It includes registration via @mcp.tool() decorator, input validation, API client instantiation, DNS record fetching from DNSDumpster, and JSON response formatting.
  @mcp.tool() async def query_domain(domain: str, ctx: Context) -> str: """Query DNSDumpster for all DNS records related to a domain. Args: domain: The domain name to query (e.g., example.com) ctx: Request context Returns: JSON string containing all DNS records """ if not domain: return json.dumps({"error": "Domain is required"}) # Validate domain if not is_valid_domain(domain): return json.dumps({"error": "Invalid domain name format"}) try: api_key = os.environ.get("DNSDUMPSTER_API_KEY") if not api_key: return json.dumps( { "error": "API key not configured. Set DNSDUMPSTER_API_KEY environment variable." } ) client = DNSDumpsterClient(api_key) try: ctx.info(f"Querying DNS records for {domain}") result = await client.get_dns_records(domain) return json.dumps(result, indent=2) finally: await client.close() except Exception as e: return json.dumps({"error": str(e)})
• server.py:170-181 (helper)

  Helper function for domain name validation, used within the query_domain handler.
  def is_valid_domain(domain: str) -> bool: """Validate a domain name. Args: domain: Domain name to validate Returns: True if the domain is valid, False otherwise """ pattern = r"^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$" return bool(re.match(pattern, domain))
• server.py:95-168 (helper)

  Main helper class implementing API client with rate limiting, caching, and the get_dns_records method directly used by the query_domain tool.
  class DNSDumpsterClient: """Client for the DNSDumpster API.""" def __init__(self, api_key: str): """Initialize the DNSDumpster API client. Args: api_key: DNSDumpster API key """ self.api_key = api_key self.api_base_url = "https://api.dnsdumpster.com/domain" self.rate_limiter = APIRateLimiter() self.cache = DNSCache() self.client = httpx.AsyncClient(timeout=30.0, headers={"X-API-Key": api_key}) async def get_dns_records(self, domain: str, page: Optional[int] = None) -> DNSData: """Query the DNSDumpster API for a domain's DNS records. Args: domain: Domain name to query page: Page number for pagination (Plus accounts only) Returns: Dictionary containing DNS records """ # Check cache first cache_key = f"{domain}:{page or 1}" cached_data = await self.cache.get(cache_key) if cached_data: return cached_data # Wait for rate limiting await self.rate_limiter.wait_for_rate_limit() # Build URL with query parameters url = f"{self.api_base_url}/{domain}" params = {} if page is not None: params["page"] = str(page) # Retry logic for network errors max_retries = 3 retry_delay = 2.0 for attempt in range(max_retries): try: response = await self.client.get(url, params=params) if response.status_code == 429: # Handle rate limiting retry_after = int(response.headers.get("Retry-After", "5")) await asyncio.sleep(retry_after) continue response.raise_for_status() data = response.json() # Cache the response await self.cache.set(cache_key, data) return data except httpx.HTTPError as e: if attempt == max_retries - 1: raise Exception(f"Failed to query DNSDumpster API: {str(e)}") # Exponential backoff await asyncio.sleep(retry_delay * (2**attempt)) async def close(self): """Close the HTTP client.""" await self.client.aclose()