DNSDumpster MCP Server

Instructions

Implementation Reference

• server.py:190-228 (handler)

  The handler function for the 'query_domain' tool. It handles input validation, API key retrieval, DNSDumpsterClient initialization, API call to fetch DNS records, and returns the result as JSON. Includes the @mcp.tool() decorator which registers the tool.

  @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:95-168 (helper)

  Core helper class DNSDumpsterClient used by query_domain. Its get_dns_records method performs the actual API query with caching and rate limiting.

  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()

• server.py:170-181 (helper)

  Helper function for domain validation used in 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:184-187 (registration)

  Creation of the FastMCP instance 'mcp' to which tools are registered via decorators.

  mcp = FastMCP(
      "mcp-dnsdumpster",
      dependencies=["httpx"],
  )