Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/threat_profiles.py:102-144 (handler)

  Main handler function that executes the get_threat_profile_recommendations tool. It fetches recommendation relationships for a given threat profile and returns sanitized results.

  @server.tool()
  async def get_threat_profile_recommendations(
      profile_id: str, ctx: Context, limit: int = 10, api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Returns the list of objects associated to a given Threat Profile.
  
    Each of these objects has one of the following types:
    - Threat Actors
    - Malware Families
    - Software or Toolkits
    - Campaigns
    - IoC Collections
    - Reports
    - Vulnerabilities
  
    We can distinguish between two other types of objects based on how
    they were associated with the Threat Profile:
  
    - **Recommended objects** are automatically recommended or assigned to
      a Threat Profile based on our proprietary ML that takes into account
      the Threat Profile's configured interests such as the targeted industries,
      target regions, source regions, malware roles and actor motivations
      to recommend the most relevant threats. These objects are identified
      by the presence of "source": "SOURCE_RECOMMENDATION"
      within the "context_attributes" response parameter below.
    - **Added objects** are assigned or added by users to a Threat Profile,
      when users find other relevant threats not automatically recommended
      by our ML module. These objects are identified by the presence of
      "source": "SOURCE_DIRECT_FOLLOW" within the "context_attributes"
      response parameter below.
  
      Args:
        profile_id (str): Threat Profile identifier at Google Threat Intelligence.
        limit: Limit the number of objects to retrieve. 10 by default.
  
      Returns:
        List of Threat (collection) objects identifiers associated to 
        the Threat Profile. Use `get_collection_report` to retrieve the full objects.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object_relationships(
          client, "threat_profiles", profile_id, ['recommendations'], limit=limit)
    return utils.sanitize_response(res.get('recommendations', []))

• gti_mcp/tools/__init__.py:14-19 (registration)

  Tool registration through wildcard import of threat_profiles module which contains the get_threat_profile_recommendations function decorated with @server.tool().

  from .collections import *
  from .files import *
  from .intelligence import *
  from .netloc import *
  from .threat_profiles import *
  from .urls import *

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

  Helper function that fetches relationship objects from Google Threat Intelligence API. Used by get_threat_profile_recommendations to retrieve the 'recommendations' relationship for threat profiles.

  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 and lists from API responses. Used to clean up the threat profile recommendations response before returning it.

  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

• gti_mcp/server.py:72-74 (registration)

  Final registration point where all tools including get_threat_profile_recommendations are imported into the MCP server, making them available through the FastMCP framework.

  # Load tools.
  from gti_mcp.tools import *