Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:546-644 (handler)

  Main implementation of get_collection_feature_matches tool. This is decorated with @server.tool() which registers it as an MCP tool. It retrieves IOCs from a collection that match a specific feature by making an API call to /collections/{collection_id}/features/search endpoint.

  @server.tool()
  async def get_collection_feature_matches(
      collection_id: str,
      feature_type: str,
      feature_id: str,
      entity_type: str,
      search_space: str,
      ctx: Context,
      descriptors_only: bool=True,
      api_key: str = None,
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Retrieves Indicators of Compromise (IOCs) from a collection that match a specific feature.
  
    This tool allows pivoting from a commonality to the specific IOCs within a collection that exhibit that feature.
    Commonalities are shared characteristics and hidden relationships between various Indicators of Compromise (e.g., files, URLs, domains, IPs).
  
    Available feature types by entity type:
    Files:
      - android_certificates, android_main_activities, android_package_names, attributions, behash,
        collections, compressed_parents, contacted_domains, contacted_ips, contacted_urls,
        crowdsourced_ids_results, crowdsourced_yara_results, elfhash, email_parents,
        embedded_domains, embedded_ips, embedded_urls, execution_parents, imphash,
        itw_domains, itw_urls, mutexes_created, mutexes_opened, pcap_parents,
        registry_keys_deleted, registry_keys_opened, registry_keys_set, tags, vhash, file_types,
        crowdsourced_sigma_results, deb_info_packages, debug_codeview_guids, debug_codeview_names,
        debug_timestamps, dropped_files_path, dropped_files_sha256, elfinfo_exports,
        elfinfo_imports, exiftool_authors, exiftool_companies, exiftool_create_dates,
        exiftool_creators, exiftool_last_modified, exiftool_last_printed, exiftool_producers,
        exiftool_subjects, exiftool_titles, filecondis_dhash, main_icon_dhash,
        main_icon_raw_md5, netassembly_mvid, nsrl_info_filenames, office_application_names,
        office_authors, office_creation_datetimes, office_last_saved, office_macro_names,
        permhash, pe_info_imports, pe_info_exports, pe_info_section_md5,
        pe_info_section_names, pwdinfo_values, sandbox_verdicts, signature_info_comments,
        signature_info_copyrights, signature_info_descriptions, signature_info_identifiers,
        signature_info_internal_names, signature_info_original_names, signature_info_products,
        symhash, trusted_verdict_filenames, rich_pe_header_hash, telfhash, tlshhash,
        email_senders, email_subjects, popular_threat_category, popular_threat_name,
        suggested_threat_label, attack_techniques, malware_config_family_name,
        malware_config_campaign_id, malware_config_campaign_group, malware_config_dga_seed,
        malware_config_dns_server, malware_config_service, malware_config_registry_key,
        malware_config_event, malware_config_pipe, malware_config_mutex, malware_config_folder,
        malware_config_file, malware_config_process_inject_target, malware_config_crypto_key,
        malware_config_displayed_message, malware_config_c2_url, malware_config_download_url,
        malware_config_misc_url, malware_config_decoy_url, malware_config_c2_user_agent,
        malware_config_download_user_agent, malware_config_misc_user_agent,
        malware_config_decoy_user_agent, malware_config_c2_password,
        malware_config_misc_username, malware_config_misc_password,
        malware_config_host_port, malware_config_dropped_file,
        malware_config_dropped_file_path, malware_config_registry_value,
        malware_config_download_password, malware_config_c2_username,
        malware_config_download_username, malware_config_exfiltration_username,
        malware_config_exfiltration_password, malware_config_exfiltration_url,
        malware_config_exfiltration_user_agent, malware_config_pivot_hash,
        memory_pattern_urls
  
    Domains:
      - attributions, collections, communicating_files, downloaded_files,
        favicon_dhash, favicon_raw_md5, urls, registrant_names
  
    IP Addresses:
      - attributions, collections, communicating_files, downloaded_files, urls
  
    URLs:
      - attributions, http_response_contents, collections, contacted_domains,
        communicating_files, cookie_names, cookie_values, downloaded_files,
        domains, embedded_js, favicon_dhash, favicon_raw_md5, html_titles,
        ip_addresses, memory_patterns, outgoing_links, path, prefix_paths,
        suffix_paths, ports, users, passwords, user_passwords, query_strings,
        query_param_keys, query_param_values, query_param_key_values,
        referring_files, tags, tracker_ids
  
    Args:
      collection_id (required): The ID of the collection to search within.
      feature_type (required): The type of feature to search for (e.g., 'attack_techniques').
      feature_id (required): The specific value of the feature (e.g., 'T1497.001').
      entity_type (required): 
      search_space (required): The scope of the search. Use 'collection' to search only within the specified collection, or 'corpus' to search across the entire VirusTotal dataset.
      entity_type_plural (required): The plural of 'entity_type'.
      descriptors_only (optional): Returns only the descriptors.
    Returns:
      A dictionary containing the list of matching IOCs.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      params = {
          "feature_type": feature_type,
          "feature_id": feature_id,
          "entity_type": entity_type,
          "search_space": search_space,
          "type": entity_type_plural,
          "descriptors_only": str(descriptors_only).lower(),
      }
      
      response = await client.get_async(f"/collections/{collection_id}/features/search", params=params)
      if response.status != 200:
          error_json = await response.json_async()
          error_info = error_json.get("error", {})
          return [{"error": f"API Error: {error_info.get('message', 'Unknown error')}"}]
      data = await response.json_async()
      return utils.sanitize_response(data.get("data", []))

• gti_mcp/tools/collections.py:546-556 (registration)

  The tool is registered via the @server.tool() decorator. This decorator registers the function as an MCP tool with the FastMCP server instance defined in gti_mcp/server.py.

  @server.tool()
  async def get_collection_feature_matches(
      collection_id: str,
      feature_type: str,
      feature_id: str,
      entity_type: str,
      search_space: str,
      ctx: Context,
      descriptors_only: bool=True,
      api_key: str = None,
  ) -> typing.List[typing.Dict[str, typing.Any]]:

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

  The vt_client context manager helper function used by get_collection_feature_matches to create and manage VirusTotal API client instances. It ensures proper cleanup of client resources.

  @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/utils.py:119-138 (helper)

  The sanitize_response helper function used by get_collection_feature_matches to recursively remove empty dictionaries and lists from API responses before returning them.

  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