Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:647-660 (handler)

  Main handler function for get_collections_commonalities tool. This async function retrieves common characteristics or features of indicators of compromise (IoC) within a collection by calling the VirusTotal API endpoint /collections/{collection_id}?attributes=aggregations and returns a markdown-formatted string of the commonalities.

  @server.tool()
  async def get_collections_commonalities(collection_id: str, ctx: Context, api_key: str = None) -> str:
    """Retrieve the common characteristics or features (attributes / relationships) of the indicators of compromise (IoC) within a collection, identified by its ID.
    Args:
      collection_id (required): Collection identifier.
    Returns:
      Markdown-formatted string with the commonalities of the collection.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      data = await client.get_async(f"/collections/{collection_id}?attributes=aggregations")
      data = await data.json_async()
      sanitized_data = utils.sanitize_response(data["data"])
      markdown_output = utils.parse_collection_commonalities(sanitized_data)
    return markdown_output

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

  Tool registration via @server.tool() decorator at line 647, which registers get_collections_commonalities as an available MCP tool.

  @server.tool()

• gti_mcp/utils.py:141-174 (helper)

  Helper function parse_collection_commonalities that converts the aggregations data from the API response into a formatted markdown string. It iterates through IOC types and their features, formatting counts, values, and prevalence information into a readable markdown format.

  def parse_collection_commonalities(data: dict) -> str:
      """
      Converts a dictionary from a JSON file to a markdown string.
      """
      markdown_string = ""
      collection_id = data.get("id", "N/A")
      markdown_string += f"# Commonalities for {collection_id}\n\n"
  
      aggregations = data.get("attributes", {}).get("aggregations", {})
      for ioc_type, features in aggregations.items():
          # Replace underscores in ioc_type
          formatted_ioc_type = ioc_type.replace('_', ' ')
          markdown_string += f"## {formatted_ioc_type} commonalities\n\n"
          
          for feature_type, feature_list in features.items():
              if isinstance(feature_list, list):
                  # Replace underscores in feature_type
                  formatted_feature_type = feature_type.replace('_', ' ')
                  markdown_string += f"### {formatted_feature_type}\n"
                  
                  for item in feature_list:
                      value = item.get("value", "N/A")
                      if isinstance(value, dict):
                          value = value.get("id", "N/A")
                      count = item.get("count", "N/A")
                      prevalence = item.get("prevalence", "N/A")
                      
                      if prevalence != "N/A" and float(prevalence) != 0:
                          markdown_string += f"- {count} matches of {value} with a prevalence of {prevalence:.8g}\n"
                      else:
                          markdown_string += f"- {count} matches of {value}\n"
                  markdown_string += "\n"
  
      return markdown_string