DNSDumpster MCP Server

Instructions

Implementation Reference

• server.py:561-646 (handler)

  Primary implementation of the search_subdomains tool. Registered with @mcp.tool() decorator. Fetches DNS records using DNSDumpsterClient and extracts unique subdomains from A and CNAME records, with pagination support.
  @mcp.tool() async def search_subdomains(domain: str, ctx: Context, page: int = 1) -> str: """Search for subdomains of a given domain. Args: domain: The parent domain name to query (e.g., example.com) ctx: Request context page: Page number for pagination (Plus accounts only) Returns: Formatted string containing subdomains found """ if not domain: return "Error: Domain is required" # Validate domain if not is_valid_domain(domain): return "Error: Invalid domain name format" try: api_key = os.environ.get("DNSDUMPSTER_API_KEY") if not api_key: return "Error: API key not configured. Set DNSDUMPSTER_API_KEY environment variable." client = DNSDumpsterClient(api_key) try: ctx.info(f"Searching subdomains for {domain} (page {page})") result = await client.get_dns_records(domain, page=page) # Extract subdomains from A records subdomains = set() if "a" in result: for record in result["a"]: host = record.get("host", "").lower() if ( host and host.endswith(domain.lower()) and host != domain.lower() ): subdomains.add(host) # Extract subdomains from CNAME records if "cname" in result: for record in result["cname"]: host = record.get("host", "").lower() if ( host and host.endswith(domain.lower()) and host != domain.lower() ): subdomains.add(host) target = record.get("target", "").lower() if ( target and target.endswith(domain.lower()) and target != domain.lower() ): subdomains.add(target) if not subdomains: return f"No subdomains found for {domain} on page {page}" output_lines = [f"Subdomains for {domain} (page {page}):"] for subdomain in sorted(subdomains): output_lines.append(f"\n{subdomain}") # Add pagination hint total_records = result.get("total_a_recs", 0) if total_records > 50 and len(subdomains) >= 50: # Free tier limit output_lines.append( f"\n\nShowing {len(subdomains)} subdomains. There may be more results available." ) output_lines.append( f"To see more results, use page parameter (e.g., page=2)" ) return "\n".join(output_lines) finally: await client.close() except Exception as e: return f"Error: {str(e)}"
• server.py:170-181 (helper)

  Helper function used by search_subdomains to validate the input domain format using regex.
  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)

  Core helper class that handles API communication with DNSDumpster, including caching, rate limiting, retries, and the get_dns_records method called by the 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()