Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/urls.py:93-159 (handler)

  Main handler function that retrieves entities related to a URL using the Google Threat Intelligence API. Validates the relationship name, converts URL to base64, fetches relationships, and sanitizes the response.

  @server.tool()
  async def get_entities_related_to_an_url(
      url: 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 URL.
  
      The following table shows a summary of available relationships for URL objects.
  
      | Relationship            | Description                                                    | Return type  |
      | ----------------------- | -------------------------------------------------------------- | ------------ | 
      | analyses                | Analyses for the URL.                                          | analyse      |
      | associations            | URL'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 URL.                               | collection   |
      | collections             | IoC Collections associated to the URL.                         | collection   |
      | comments                | Community posted comments about the URL.                       | comment      |
      | communicating_files     | Files that communicate with a given URL when they're executed. | file         |
      | contacted_domains       | Domains from which the URL loads some kind of resource.        | domain       |
      | contacted_ips           | IPs from which the URL loads some kind of resource.            | ip_address   |
      | downloaded_files        | Files downloaded from the URL.                                 | file         |
      | embedded_js_files       | JS files embedded in a URL.                                    | file         |
      | graphs                  | Graphs including the URL.                                      | graph        |
      | http_response_contents  | HTTP response contents from the URL.                           | file         |
      | last_serving_ip_address | Last IP address that served the URL.                           | ip_address   |
      | malware_families        | Malware families associated to the URL.                        | collection   |
      | memory_pattern_parents  | Files having a domain as string on memory during sandbox execution. | file    |
      | network_location        | Domain or IP for the URL.                                      | domain or ip_address |
      | parent_resource_urls    | Returns the URLs where this URL has been loaded as resource.   | url          |
      | redirecting_urls        | URLs that redirected to the given URL.                         | url          |
      | redirects_to            | URLs that this url redirects to.                               | url          |
      | referrer_files          | Files containing the URL.                                      | file         |
      | referrer_urls           | URLs referring the URL.                                        | url          |
      | related_collections     | Returns the Collections of the parent Domains or IPs of this URL. | collection  |
      | related_comments        | Community posted comments in the URL's related objects.        | comment      |
      | related_reports         | Reports that are directly and indirectly related to the URL.   | collection   |
      | related_threat_actors   | URL's related threat actors.                                   | collection   |
      | reports                 | Reports directly associated to the URL.                        | collection   |
      | software_toolkits       | Software and Toolkits associated to the URL.                   | collection   |
      | submissions             | URL's submissions.                                             | url          |
      | urls_related_by_tracker_id | URLs that share the same tracker ID.                        | url          |
      | user_votes          | URL's votes made by current signed-in user.                        | vote         |
      | votes                  | Votes for the URL.                                              | vote         |
      | vulnerabilities        | Vulnerabilities associated to the URL.                          | collection   |
  
      Args:
        url (required): URL 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 objects to retrieve. 10 by default.
      Returns:
        List of entities related to the URL.
    """
    if not relationship_name in URL_RELATIONSHIPS:
      return {
         "error": f"Relationship {relationship_name} does not exist. "
                  f"Available relationships are: {','.join(URL_RELATIONSHIPS)}"
      }
  
    url_id = url_to_base64(url)
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object_relationships(
          client,
          "urls", 
          url_id,
          relationships=[relationship_name],
          descriptors_only=descriptors_only,
          limit=limit)
    return utils.sanitize_response(res.get(relationship_name, []))

• gti_mcp/tools/urls.py:93-93 (registration)

  The @server.tool() decorator that registers the get_entities_related_to_an_url function as an MCP tool with the FastMCP server.

  @server.tool()

• gti_mcp/tools/urls.py:63-69 (helper)

  Helper function that converts a URL to base64 encoding without padding, as required by the Google Threat Intelligence API for URL identifiers.

  def url_to_base64(url: str) -> str:
    """Converts the URL into base64.
  
    Without padding, as required by the Google Threat Intelligence API.
    """
    b = base64.b64encode(url.encode('utf-8'))
    return b.decode('utf-8').rstrip("=")

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

  Helper function that asynchronously fetches relationship data from the VirusTotal API for a given object, using TaskGroup for concurrent requests and optionally returning descriptors only.

  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)

  Helper function that recursively removes empty dictionaries, lists, and empty strings from API response data 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