Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/files.py:110-190 (handler)

  The main handler function that retrieves entities related to a file hash using VirusTotal API relationships

  @server.tool()
  async def get_entities_related_to_a_file(
      hash: 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 file hash.
  
      The following table shows a summary of available relationships for file objects.
  
      | Relationship           | Description                                                                       | Return type |
      | ---------------------- | --------------------------------------------------------------------------------- | ----------- |
      | analyses               | Analyses for the file                                                             | analysis |
      | associations           | File's associated objects (reports, campaigns, IoC collections, malware families, software toolkits, vulnerabilities, threat-actors), without filtering by the associated object type.                                                                                      | collection |
      | behaviours             | Behaviour reports for the file.                                                   | file-behaviour |
      | attack_techniques      | Returns the Attack Techniques of the File.                                        | attack_technique |
      | bundled_files          | Files bundled within the file.                                                    | file |
      | campaigns              | Campaigns associated to the file.                                                 | collection |
      | carbonblack_children   | Files derived from the file according to Carbon Black.                            | file |
      | carbonblack_parents    | Files from where the file was derived according to Carbon Black.                  | file |
      | collections            | IoC Collections associated to the file.                                           | collection |
      | comments               | Comments for the file.                                                            | comment |
      | compressed_parents     | Compressed files that contain the file.                                           | file |
      | contacted_domains      | Domains contacted by the file.                                                    | domain |
      | contacted_ips          | IP addresses contacted by the file.                                               | ip_address |
      | contacted_urls         | URLs contacted by the file.                                                       | url |
      | dropped_files          | Files dropped by the file during its execution.                                   | file |
      | email_attachments      | Files attached to the email.                                                      | file |
      | email_parents          | Email files that contained the file.                                              | file |
      | embedded_domains       | Domain names embedded in the file.                                                | domain |
      | embedded_ips           | IP addresses embedded in the file.                                                | ip_address |
      | embedded_urls          | URLs embedded in the file.                                                        | url |
      | execution_parents      | Files that executed the file.                                                     | file |
      | graphs                 | Graphs that include the file.                                                     | graph |
      | itw_domains            | In the wild domain names from where the file has been downloaded.                 | domain |
      | itw_ips                | In the wild IP addresses from where the file has been downloaded.                 | ip_address |
      | itw_urls               | In the wild URLs from where the file has been downloaded.                         | url |
      | malware_families       | Malware families associated to the file.                                          | collection |
      | memory_pattern_domains | Domain string patterns found in memory during sandbox execution.                  | domain |
      | memory_pattern_ips     | IP address string patterns found in memory during sandbox execution.              | ip_address |
      | memory_pattern_urls    | URL string patterns found in memory during sandbox execution.                     | url |
      | overlay_children       | Files contained by the file as an overlay.                                        | file |
      | overlay_parents        | File that contain the file as an overlay.                                         | file |
      | pcap_children          | Files contained within the PCAP file.                                             | file |
      | pcap_parents           | PCAP files that contain the file.                                                 | file |
      | pe_resource_children   | Files contained by a PE file as a resource.                                       | file |
      | pe_resource_parents    | PE files containing the file as a resource.                                       | file |
      | related_attack_techniques    | Returns the Attack Techniques of the Collections containing this File.      | attack_technique |
      | related_reports        | Reports that are directly and indirectly related to the file.                     | collection |
      | related_threat_actors  | File's related threat actors.                                                     | collection |
      | reports                | Reports directly associated to the file.                                          | collection |
      | screenshots            | Screenshots related to the sandbox execution of the file.                         | screenshot |
      | similar_files          | Files that are similar to the file.                                               | file |
      | software_toolkits      | Software and Toolkits associated to the file.                                     | collection |
      | submissions            | Submissions for the file.                                                         | submission |
      | urls_for_embedded_js   | URLs where this (JS) file is embedded.                                            | url |
      | user_votes             | File's votes made by current signed-in user.                                      | vote |
      | votes                  | Votes for the file.                                                               | vote |
      | vulnerabilities        | Vulnerabilities associated to the file.                                           | collection |
  
      Args:
        hash (required): MD5/SHA1/SHA256) hash that identifies the file.
        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 files to retrieve. 10 by default.
      Returns:
        List of objects related to the given file.
      """
      if not relationship_name in FILE_RELATIONSHIPS:
          return {
              "error": f"Relationship {relationship_name} does not exist. "
              f"Available relationships are: {','.join(FILE_RELATIONSHIPS)}"
          }
  
      async with vt_client(ctx, api_key=api_key) as client:
        res = await utils.fetch_object_relationships(
            client, 
            "files",
            hash, 
            relationships=[relationship_name],
            descriptors_only=descriptors_only,
            limit=limit)
      return utils.sanitize_response(res.get(relationship_name, []))

• gti_mcp/tools/files.py:110-111 (registration)

  Registration of the tool using the @server.tool() decorator from FastMCP framework

  @server.tool()
  async def get_entities_related_to_a_file(

• gti_mcp/tools/files.py:26-74 (schema)

  Schema definition of all valid relationship types that can be queried for file objects

  FILE_RELATIONSHIPS = [
      "analyses",
      "associations",
      "behaviours",
      "attack_techniques",
      "bundled_files",
      "campaigns",
      "carbonblack_children",
      "carbonblack_parents",
      "collections",
      "comments",
      "compressed_parents",
      "contacted_domains",
      "contacted_ips",
      "contacted_urls",
      "dropped_files",
      "email_attachments",
      "email_parents",
      "embedded_domains",
      "embedded_ips",
      "embedded_urls",
      "execution_parents",
      "graphs",
      "itw_domains",
      "itw_ips",
      "itw_urls",
      "malware_families",
      "memory_pattern_domains",
      "memory_pattern_ips",
      "memory_pattern_urls",
      "overlay_children",
      "overlay_parents",
      "pcap_children",
      "pcap_parents",
      "pe_resource_children",
      "pe_resource_parents",
      "related_attack_techniques",
      "related_reports",
      "related_threat_actors",
      "reports",
      "screenshots",
      "similar_files",
      "software_toolkits",
      "submissions",
      "urls_for_embedded_js",
      "user_votes",
      "votes",
      "vulnerabilities",
  ]

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

  Helper function that fetches relationship data from VirusTotal API using async iterators and task groups

  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 recursively removes empty dictionaries and lists from API responses to clean up the output

  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