Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/netloc.py:196-251 (handler)

  Main handler function get_entities_related_to_an_ip_address that retrieves entities related to an IP address. Validates the relationship_name parameter against IP_RELATIONSHIPS, calls the VirusTotal API via fetch_object_relationships, and sanitizes the response.

  @server.tool()
  async def get_entities_related_to_an_ip_address(
      ip_address: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10, api_key: str = None
  ) -> list[dict[str, typing.Any]]:
    """Retrieve entities related to the the given IP Address.
  
      The following table shows a summary of available relationships for IP Address objects.
  
      | Relationship                | Description                                            | Return type  |
      | --------------------------- | ------------------------------------------------------ | ------------ |
      | associations                | IP's associated objects (reports, campaigns, IoC collections, malware families, software toolkits, vulnerabilities, threat-actors), without filtering by the associated object type. | collection |
      | campaigns                   | Campaigns associated to the IP address.                | collection   |
      | collections                 | IoC Collections associated to the IP address.          | collection   |
      | comments                    | Comments for the IP address.                           | comment      |
      | communicating_files         | Files that communicate with the IP address.            | file         |
      | downloaded_files            | Files downloaded from the IP address.                  | file         |
      | graphs                      | Graphs including the IP address.                       | graph        |
      | historical_ssl_certificates | SSL certificates associated with the IP.               | ssl-cert     |
      | historical_whois            | WHOIS information for the IP address.                  | whois        |
      | malware_families            | Malware families associated to the IP address.         | collection   |
      | memory_pattern_parents      | Files having an IP as string on memory during sandbox execution. | file |
      | referrer_files              | Files containing the IP address.                       | file         |
      | related_comments            | Community posted comments in the IP's related objects. | comment      |
      | related_reports             | Reports that are directly and indirectly related to the IP. | collection |
      | related_threat_actors       | Threat actors related to the IP address.               | collection   |
      | reports                     | Reports directly associated to the IP.                 | collection   |
      | resolutions                 | IP address' resolutions                                | resolution   |
      | software_toolkits           | Software and Toolkits associated to the IP address.    | collection   |
      | urls                        | URLs related to the IP address.                        | url          |
      | user_votes                  | IP's votes made by current signed-in user.             | vote         |
      | votes                       | IP's votes.                                            | vote         |
      | vulnerabilities             | Vulnerabilities associated to the IP address.          | collection   |
  
      Args:
        ip_address (required): IP Addres to analyse.
        relationship_name (required): Relationship name.
        descriptors_only (required): Bool. Must be True when the target object type is one of file, domain, url, ip_address or collection.
        limit: Limit the number of entities to retrieve. 10 by default.
      Returns:
        List of entities related to the IP Address.
    """
    if not relationship_name in IP_RELATIONSHIPS:
      return {
         "error": f"Relationship {relationship_name} does not exist. "
                  f"Available relationships are: {','.join(IP_RELATIONSHIPS)}"
      }
  
    async with vt_client(ctx) as client:
      res = await utils.fetch_object_relationships(
          client, 
          "ip_addresses",
          ip_address,
          relationships=[relationship_name],
          descriptors_only=descriptors_only,
          limit=limit)
    return utils.sanitize_response(res.get(relationship_name, []))

• gti_mcp/tools/netloc.py:61-84 (schema)

  IP_RELATIONSHIPS constant defining all valid relationship names that can be queried for IP addresses. This serves as validation schema for the relationship_name parameter.

  IP_RELATIONSHIPS = [
      "associations",
      "campaigns",
      "collections",
      "comments",
      "communicating_files",
      "downloaded_files",
      "graphs",
      "historical_ssl_certificates",
      "historical_whois",
      "malware_families",
      "memory_pattern_parents",
      "referrer_files",
      "related_comments",
      "related_reports",
      "related_threat_actors",
      "reports",
      "resolutions",
      "software_toolkits",
      "urls",
      "user_votes",
      "votes",
      "vulnerabilities",
  ]

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

  fetch_object_relationships helper function that fetches relationship descriptors from VirusTotal objects. Creates async tasks to fetch multiple relationships concurrently and aggregates the results.

  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

• gti_mcp/utils.py:119-138 (helper)

  sanitize_response helper function that recursively removes empty dictionaries and lists from API responses to clean up the output.

  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