Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:86-138 (handler)

  Main handler function for get_entities_related_to_a_collection tool. Includes @server.tool() registration decorators, function signature with parameters (id, relationship_name, ctx, limit, descriptors_only, api_key), comprehensive docstring documenting available relationships and their types, input validation checking if relationship_name is in COLLECTION_RELATIONSHIPS, and the core logic that calls utils.fetch_object_relationships and returns sanitized results.

  @server.tool()
  @server.tool()
  async def get_entities_related_to_a_collection(
      id: str, relationship_name: str, ctx: Context, limit: int = 10, descriptors_only: bool = True, api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Retrieve entities related to the the given collection ID.
  
      The following table shows a summary of available relationships for collection objects.
  
      | Relationship         | Description                                       | Return type  |
      | -------------------- | ------------------------------------------------- | ------------ |
      | associations         | List of associated threats                        | collection   |
      | attack_techniques    | List of attack techniques                         | attack_technique |
      | domains              | List of Domains                                   | domain       |
      | files                | List of Files                                     | file         |
      | ip_addresses         | List of IP addresses                              | ip_address   |
      | urls                 | List of URLs                                      | url          |
      | threat_actors        | List of related threat actors                     | collection   |
      | malware_families     | List of related malware families                  | collection   |
      | software_toolkits    | List of related tools                             | collection   |
      | campaigns            | List of related campaigns                         | collection   |
      | vulnerabilities      | List of related vulnerabilities                   | collection   |
      | reports              | List of reports                                   | collection   |
      | suspected_threat_actors | List of related suspected threat actors        | collection   |
      | hunting_rulesets     | Google Threat Intelligence Yara rules that identify the given collection | hunting_ruleset |
  
      Note on descriptors_only: When True, returns basic descriptors. When False, returns
      detailed attributes.
      IMPORTANT: `descriptors_only` must be `False` for the 'attack_techniques' relationship.
      
      Args:
        id (required): Collection identifier.
        relationship_name (required): Relationship name.
        limit (optional): Limit the number of collections to retrieve. 10 by default.
        descriptors_only (optional)): Bool. Default True. Must be False when the target object type is 'attack_techniques'.
      Returns:
        List of objects related to the collection.
    """
    if not relationship_name in COLLECTION_RELATIONSHIPS:
        return {
            "error": f"Relationship {relationship_name} does not exist. "
            f"Available relationships are: {','.join(COLLECTION_RELATIONSHIPS)}"
        }
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object_relationships(
          client, 
          "collections", 
          id, 
          [relationship_name],
          descriptors_only=descriptors_only,
          limit=limit)
    return utils.sanitize_response(res.get(relationship_name, []))

• gti_mcp/tools/collections.py:24-39 (schema)

  Schema constant defining valid relationship names for collection objects. This list is used for input validation in get_entities_related_to_a_collection. Includes relationships like associations, attack_techniques, domains, files, ip_addresses, urls, threat_actors, malware_families, software_toolkits, campaigns, vulnerabilities, reports, suspected_threat_actors, and hunting_rulesets.

  COLLECTION_RELATIONSHIPS = [
      "associations",
      "attack_techniques",
      "domains",
      "files",
      "ip_addresses",
      "urls",
      "threat_actors",
      "malware_families",
      "software_toolkits",
      "campaigns",
      "vulnerabilities",
      "reports",
      "suspected_threat_actors",
      "hunting_rulesets",
  ]

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

  Registration decorators that register get_entities_related_to_a_collection as an MCP tool. The @server.tool() decorator is applied twice (lines 86-87) to expose this function as a tool in the MCP server.

  @server.tool()
  @server.tool()

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

  Helper function fetch_object_relationships that handles the actual API call to retrieve relationships from VirusTotal. It constructs the API endpoint based on whether descriptors_only is True or False, uses asyncio.TaskGroup to fetch relationships concurrently, consumes the VT iterator, removes aggregations from attributes, and returns the data in a structured format.

  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