Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:242-263 (handler)

  The search_campaigns tool handler - decorated with @server.tool() and implements the search logic for threat campaigns by calling _search_threats_by_collection_type with 'campaign' as the collection type

  @server.tool()
  @server.tool()
  async def search_campaigns(
      query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-", api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Search threat campaigns in the Google Threat Intelligence platform.
  
    Campaigns are modeled as collections. Once you get collections from this tool, you can use `get_collection_report` to fetch the full reports and their relationships.
  
    You can use order_by to sort the results by: "relevance", "creation_date". You can use the sign "+" to make it order ascending, or "-" to make it descending. By default is "relevance-"
  
    Args:
      query (required): Search query to find threats.
      limit: Limit the number of threats to retrieve. 10 by default.
      order_by: Order results by the given order key. "relevance-" by default.
  
    Returns:
      List of collections, aka threats.
    """
    res = await _search_threats_by_collection_type(
        query, "campaign", ctx, limit, order_by, api_key=api_key)
    return res

• gti_mcp/tools/collections.py:140-175 (helper)

  Helper function _search_threats_by_collection_type that performs the actual API call to the Google Threat Intelligence platform /collections endpoint with filtering by collection_type and sanitizes the response

  async def _search_threats_by_collection_type(
      query: str,
      collection_type: str,
      ctx: Context,
      limit: int = 10,
      order_by: str = "relevance-",
      api_key: str = None,
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Search a given threat type in the Google Threat Intelligence platform,
  
    Args:
      query (required): Search query to find threats. If you want any threat, just pass an empty string.
      collection_type (required): Collection type. One of: "threat-actor", "malware-family", "campaign", "report", "vulnerability", "collection".
      limit: Limit the number of threats to retrieve. 10 by default.
      order_by: Order results by the given order key. "relevance-" by default.
  
    Returns:
      List of collections, aka threats.
    """
    if collection_type not in COLLECTION_TYPES:
        raise ValueError(
            f"wrong collection_type. Available collection_type are: {','.join(COLLECTION_TYPES)} ")
  
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.consume_vt_iterator(
          client,
          "/collections",
          params={
              "filter": f"collection_type:{collection_type} {query}",
              "order": order_by,
              "relationships": COLLECTION_KEY_RELATIONSHIPS,
              "exclude_attributes": COLLECTION_EXCLUDED_ATTRS,
          },
          limit=limit,
      )
    return utils.sanitize_response([o.to_dict() for o in res])

• gti_mcp/tools/collections.py:46-54 (schema)

  COLLECTION_TYPES schema definition - a set of valid collection types that includes 'campaign' and is used for validation in the helper function

  COLLECTION_TYPES = {
      "threat-actor",
      "malware-family",
      "campaign",
      "report",
      "software-toolkit",
      "vulnerability",
      "collection",
  }