Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/netloc.py:111-174 (handler)

  The main handler function for 'get_entities_related_to_a_domain' tool. This function retrieves entities related to a given domain by calling the VirusTotal API via fetch_object_relationships helper. It validates the relationship_name parameter against DOMAIN_RELATIONSHIPS, then returns a sanitized list of related entities.

  @server.tool()
  async def get_entities_related_to_a_domain(
      domain: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10, api_key: str = None
  ) -> list[dict[str, typing.Any]]:
    """Retrieve entities related to the the given domain.
  
      The following table shows a summary of available relationships for domain objects.
  
      | Relationship                | Description                                                | Return type  |
      | --------------------------- | ---------------------------------------------------------- | ------------ |
      | associations                | Domain's associated objects (reports, campaigns, IoC collections, malware families, software toolkits, vulnerabilities, threat-actors), without filtering by the associated object type.                                                             | Everyone. | List of [reports](ref:report-object), [campaigns](ref:campaign-object), [IoC collections](ref:ioc-collection-object), [malware families](ref:malware-family-object), [software toolkits](ref:software-toolkit-object), [vulnerabilities](ref:vulnerability-object), [threat-actors](ref:threat-actor-object) objecs.| collection |
      | caa_records                 | Records CAA for the domain.                                | domain       |
      | campaigns                   | Campaigns associated to the domain.                        | collection   |
      | cname_records               | Records CNAME for the domain.                              | domain       |
      | collections                 | IoC Collections associated to the domain.                  | collection   |
      | comments                    | Community posted comments about the domain.                | comment      |
      | communicating_files         | Files that communicate with the domain.                    | file         |
      | downloaded_files            | Files downloaded from that domain.                         | file         |
      | graphs                      | Graphs including the domain.                               | graph        |
      | historical_ssl_certificates | SSL certificates associated with the domain.               | ssl-cert     |
      | historical_whois            | WHOIS information for the domain.                          | whois        |
      | immediate_parent            | Domain's immediate parent.                                 | domain       |
      | malware_families            | Malware families associated to the domain.                 | collection   |
      | memory_pattern_parents      | Files having a domain as string on memory during sandbox execution. | file |
      | mx_records                  | Records MX for the domain.                                 | domain       |
      | ns_records                  | Records NS for the domain.                                 | domain       |
      | parent                      | Domain's top parent.                                       | domain       |
      | referrer_files              | Files containing the domain.                               | file         |
      | related_comments            | Community posted comments in the domain's related objects. | comment      |
      | related_reports             | Reports that are directly and indirectly related to the domain. | collection |
      | related_threat_actors       | Threat actors related to the domain.                       | collection   |
      | reports                     | Reports directly associated to the domain.                 | collection   |
      | resolutions                 | DNS resolutions for the domain.                            | resolution   |
      | siblings                    | Domain's sibling domains.                                  | domain       |
      | soa_records                 | Records SOA for the domain.                                | domain       |
      | software_toolkits           | Software and Toolkits associated to the domain.            | collection   |
      | subdomains                  | Domain's subdomains.                                       | domain       |
      | urls                        | URLs having this domain.                                   | url          |
      | user_votes                  | Current user's votes.                                      | vote         |
      | votes                       | Domain's votes.                                            | vote         |
      | vulnerabilities             | Vulnerabilities associated to the domain.                  | collection   |
  
      Args:
        domain (required): Domain to analyse.
        relationship_name (required): Relationship name.
        descriptors_only (required): Bool. Must be True when the target object type is one of file, domain, url, ip_address or collection.
        limit: Limit the number of entities to retrieve. 10 by default.
      Returns:
        List of entities related to the domain.
    """
    if not relationship_name in DOMAIN_RELATIONSHIPS:
      return {
         "error": f"Relationship {relationship_name} does not exist. "
                  f"Available relationships are: {','.join(DOMAIN_RELATIONSHIPS)}"
      }
  
    async with vt_client(ctx) as client:
      res = await utils.fetch_object_relationships(
          client, 
          "domains", domain, 
          relationships=[relationship_name],
          descriptors_only=descriptors_only,
          limit=limit)
    return utils.sanitize_response(res.get(relationship_name, []))

• gti_mcp/tools/netloc.py:22-54 (schema)

  Schema definition for valid domain relationships. This list is used to validate the relationship_name parameter in the get_entities_related_to_a_domain function.

  DOMAIN_RELATIONSHIPS = [
      "associations",
      "caa_records",
      "campaigns",
      "cname_records",
      "collections",
      "comments",
      "communicating_files",
      "downloaded_files",
      "graphs",
      "historical_ssl_certificates",
      "historical_whois",
      "immediate_parent",
      "malware_families",
      "memory_pattern_parents",
      "mx_records",
      "ns_records",
      "parent",
      "referrer_files",
      "related_comments",
      "related_reports",
      "related_threat_actors",
      "reports",
      "resolutions",
      "siblings",
      "soa_records",
      "software_toolkits",
      "subdomains",
      "urls",
      "user_votes",
      "votes",
      "vulnerabilities",
  ]

• gti_mcp/utils.py:87-116 (helper)

  Helper function that fetches object relationships from the VirusTotal API. This is called by get_entities_related_to_a_domain to retrieve the actual relationship data.

  async def fetch_object_relationships(
      vt_client: vt.Client,
      resource_collection_type: str,
      resource_id: str,
      relationships: typing.List[str],
      params: dict[str, typing.Any] | None = None,
      descriptors_only: bool = True,
      limit: int = 10):
    """Fetches the given relationships descriptors from the given object."""
    rel_futures = {}
    # If true, returns descriptors instead of full objects.
    descriptors = '/relationship' if descriptors_only else ''
    async with asyncio.TaskGroup() as tg:
      for rel_name in relationships:
        rel_futures[rel_name] = tg.create_task(
            consume_vt_iterator(
                vt_client,
                f"/{resource_collection_type}/{resource_id}"
                f"{descriptors}/{rel_name}", params=params, limit=limit))
  
    data = {}
    for name, items in rel_futures.items():
      data[name] = []
      for obj in items.result():
        obj_dict = obj.to_dict()
        if 'aggregations' in obj_dict['attributes']:
          del obj_dict['attributes']['aggregations']
        data[name].append(obj_dict)
  
    return data

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

  Helper function that sanitizes API responses by removing empty dictionaries and lists recursively. Used by get_entities_related_to_a_domain to clean up the response 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)

  Context manager that provides a VirusTotal client instance for the current request. Used by get_entities_related_to_a_domain to get an API client to fetch relationship 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()