get_domain_report

Analyze domain security risks using Google Threat Intelligence to detect threats, malware, and suspicious activity for investigation.

Instructions

Get a comprehensive domain analysis report from Google Threat Intelligence.

Args: domain (required): Domain to analyse. Returns: Report with insights about the domain.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`domain`	Yes
`api_key`	No

Implementation Reference

gti_mcp/tools/netloc.py:91-108 (handler)

Main handler function for get_domain_report tool. Decorated with @server.tool() to register it as an MCP tool. Fetches domain analysis from Google Threat Intelligence API with key relationships and returns a sanitized report.

@server.tool()
async def get_domain_report(domain: str, ctx: Context, api_key: str = None) -> typing.Dict[str, typing.Any]:
  """Get a comprehensive domain analysis report from Google Threat Intelligence.

  Args:
    domain (required): Domain to analyse.
  Returns:
    Report with insights about the domain.
  """
  async with vt_client(ctx, api_key=api_key) as client:
    res = await utils.fetch_object(
        client,
        "domains",
        "domain",
        domain,
        relationships=DOMAIN_KEY_RELATIONSHIPS,
        params={"exclude_attributes": "last_analysis_results"})
  return utils.sanitize_response(res)

gti_mcp/utils.py:29-84 (helper)

Helper function that fetches objects from Google Threat Intelligence API. Used by get_domain_report to retrieve domain data with specified attributes and relationships.

async def fetch_object(
    vt_client: vt.Client,
    resource_collection_type: str,
    resource_type: str,
    resource_id: str,
    attributes: list[str] | None = None,
    relationships: list[str] | None = None,
    params: dict[str, typing.Any] | None = None):
  """Fetches objects from Google Threat Intelligence API."""
  logging.info(
      f"Fetching comprehensive {resource_collection_type} "
      f"report for id: {resource_id}")
  
  params = {k: v for k, v in params.items()} if params else {}

  # Retrieve a selection of object attributes and/or relationships.
  if attributes:
    params["attributes"] = ",".join(attributes)
  if relationships:
    params["relationships"] = ",".join(relationships)

  try:
    obj = await vt_client.get_object_async(
        f"/{resource_collection_type}/{resource_id}", params=params)

    if obj.error:
      logging.error(
          f"Error fetching main {resource_type} report for {resource_id}: {obj.error}"
      )
      return {
          "error": f"Failed to get main {resource_type} report: {obj.error}",
          # "details": report.get("details"),
      }
  except vt.error.APIError as e:
    logging.warning(
        f"VirusTotal API Error fetching {resource_type} {resource_id}: {e.code} - {e.message}"
    )
    return {
        "error": f"VirusTotal API Error: {e.code} - {e.message}",
        "details": f"The requested {resource_type} '{resource_id}' could not be found or there was an issue with the API request."
    }
  except Exception as e:
    logging.exception(
        f"Unexpected error fetching {resource_type} {resource_id}: {e}"
    )
    return {"error": "An unexpected internal error occurred."}

  # Build response.
  obj_dict = obj.to_dict()
  obj_dict['id'] = obj.id
  if 'aggregations' in obj_dict['attributes']:
    del obj_dict['attributes']['aggregations']

  logging.info(
      f"Successfully generated concise threat summary for id: {resource_id}")
  return obj_dict

gti_mcp/utils.py:119-138 (helper)

Helper function that recursively removes empty dictionaries and lists from API responses. Used to clean up the domain report before returning it.

def sanitize_response(data: typing.Any) -> typing.Any:
  """Removes empty dictionaries and lists recursively from a response."""
  if isinstance(data, dict):
    sanitized_dict = {}
    for key, value in data.items():
      sanitized_value = sanitize_response(value)
      if sanitized_value is not None:
        sanitized_dict[key] = sanitized_value
    return sanitized_dict
  elif isinstance(data, list):
    sanitized_list = []
    for item in data:
      sanitized_item = sanitize_response(item)
      if sanitized_item is not None:
        sanitized_list.append(sanitized_item)
    return sanitized_list
  elif isinstance(data, str):
    return data if data else None
  else:
    return data

gti_mcp/server.py:56-64 (helper)

Async context manager that provides a VirusTotal client instance for API requests. Used by get_domain_report to get a client for fetching domain data.

@asynccontextmanager
async def vt_client(ctx: Context, api_key: str = None) -> AsyncIterator[vt.Client]:
  """Provides a vt.Client instance for the current request."""
  client = vt_client_factory(ctx, api_key)

  try:
    yield client
  finally:
    await client.close_async()

gti_mcp/tools/netloc.py:56-58 (helper)
Constant defining which relationships to include in domain reports. Used by get_domain_report to specify which relationships to fetch.
```
DOMAIN_KEY_RELATIONSHIPS = [
    "associations",
]
```

Google Threat Intelligence MCP Server