search_threats

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:178-239 (handler)

  The main handler function for the 'search_threats' MCP tool. It is decorated with @server.tool(), accepts a query, optional collection_type filter, limit, order_by, and api_key. It builds a filter string (optionally prepending 'collection_type:<type>' if provided) and calls utils.consume_vt_iterator to query the '/collections' endpoint of the VirusTotal/Google Threat Intelligence API.

  @server.tool()
  async def search_threats(
      ctx: Context,
      query: str,
      collection_type: str = None,
      limit: int = 5,
      order_by: str = "relevance-",
      api_key: str = None,
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Search threats in the Google Threat Intelligence platform.
  
    Threats 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.
  
    **IMPORTANT CONTEXT CLUE:** Pay close attention to the user's request. If their request mentions specific *kinds* of threats such as "threat actor", "malware family", "campaign", "report", or "vulnerability", treat this as a strong signal that you **must** use the `collection_type` filter in your `query` to ensure relevant results. Using this filter significantly improves search precision.
  
    **Filtering by Type:**
    To filter your search results to a specific *type* of threat, include the `collection_type` modifier *within your query string*.
    Syntax: `collection_type:"<type>"`
    Available `<type>` values:
      - "threat-actor": Use when the user asks about specific actors, groups, or APTs.
      - "malware-family": Use when the user asks about malware, trojans, viruses, ransomware families.
      - "software-toolkit": Use when the user asks about legit tools usually related to malware.
      - "campaign": Use when the user asks about specific attack campaigns.
      - "report": Use when the user is looking for analysis reports.
      - "vulnerability": Use when the user asks about specific CVEs or vulnerabilities.
      - "collection": A generic type, use only if no other type fits or if the user explicitly asks for generic "collections".
  
    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-"
  
    When asked for latest threats, prioritize campaigns or vulnerabilities over reports.
  
    Args:
      query (required): Search query to find threats.
      collection_type: Filter your search results to a specific *type* of threat
      limit: Limit the number of threats to retrieve. 5 by default.
      order_by: Order results by the given order key. "relevance-" by default.
  
    Returns:
      List of collections, aka threats. They are full collection objects, you do not need to retrieve them`using the `get_collection_report` tool. You may need to extend with relationships using `get_entities_related_to_a_collection` tool.
    """
    filter = ""
    if collection_type:
      if collection_type not in COLLECTION_TYPES:
        raise ValueError(
            f"unknown collection_type. Available are {','.join(COLLECTION_TYPES)}")
      filter += f"collection_type:{collection_type} "
    if query:
      filter += query
    
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.consume_vt_iterator(
          client,
          "/collections",
          params={
              "filter": filter,
              "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:24-54 (schema)

  Schema definitions used by search_threats: COLLECTION_RELATIONSHIPS (list of valid relationship names), COLLECTION_KEY_RELATIONSHIPS (just 'associations'), COLLECTION_EXCLUDED_ATTRS ('aggregations'), and COLLECTION_TYPES (set of valid collection types like 'threat-actor', 'malware-family', 'campaign', 'report', 'software-toolkit', 'vulnerability', 'collection').

  COLLECTION_RELATIONSHIPS = [
      "associations",
      "attack_techniques",
      "domains",
      "files",
      "ip_addresses",
      "urls",
      "threat_actors",
      "malware_families",
      "software_toolkits",
      "campaigns",
      "vulnerabilities",
      "reports",
      "suspected_threat_actors",
      "hunting_rulesets",
  ]
  
  COLLECTION_KEY_RELATIONSHIPS = [
      "associations",
  ]
  COLLECTION_EXCLUDED_ATTRS = ",".join(["aggregations"])
  
  COLLECTION_TYPES = {
      "threat-actor",
      "malware-family",
      "campaign",
      "report",
      "software-toolkit",
      "vulnerability",
      "collection",
  }

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

  A helper async function _search_threats_by_collection_type that searches threats filtered by a specific collection_type. Called by other specialized search tools (e.g., search_campaigns, search_threat_actors) but not directly by search_threats itself.

  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/utils.py:20-26 (helper)

  The consume_vt_iterator helper used by search_threats to iterate through paginated API results from the VirusTotal client.

  async def consume_vt_iterator(
      vt_client: vt.Client, endpoint: str, params: dict | None = None, limit: int = 10):
    """Consumes a vt.Iterator iterator and return the list of objects."""
    res = []
    async for obj in vt_client.iterator(endpoint, params=params, limit=limit):
      res.append(obj)
    return res

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

  The @server.tool() decorator on the search_threats function registers it as an MCP tool with the FastMCP server instance. The tools module is imported by gti_mcp/server.py via 'from gti_mcp.tools import *'.

  @server.tool()