Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/threat_profiles.py:54-99 (handler)

  The main implementation of the get_threat_profile tool. This async function fetches a Threat Profile object from Google Threat Intelligence API using a profile_id. It uses the vt_client context manager and utils.fetch_object helper to make the API call, then sanitizes the response.

  @server.tool()
  async def get_threat_profile(
      profile_id: str, ctx: Context, api_key: str = None
  ) -> typing.Dict[str, typing.Any]:
    """Get Threat Profile object.
  
    A threat profile object contains the following attributes:
  
      - enable_recommendations (bool): whether or not Recommendations automatically
                                        generated by our ML are enabled.
      - interests (dict): Threat Profile's configured interests such as industries, target regions,
                          source regions, malware roles and actor motivations 
                          to recommend the most relevant threats.
        - INTEREST_TYPE_TARGETED_INDUSTRY (list[str]): List of targeted industries. 
        - INTEREST_TYPE_TARGETED_REGION (list[str]): list of targeted regions (ISO-3166 country code).
        - INTEREST_TYPE_SOURCE_REGION (list[str]): list of source regions (ISO-3166 country code).
        - INTEREST_TYPE_MALWARE_ROLE (list[str]): list of malware roles. 
        - INTEREST_TYPE_ACTOR_MOTIVATION: (list[str]): list of threat actors motivations.
      - last_modification_date: <integer> Threat Profile's last modification date (UTC timestamp).
      - name (str): Threat Profile's name.
      - creation_date (int): Threat Profile's creation date (UTC timestamp).
      - aliases (list[str]): alternative names by which the threat actor is known.
      - description (str): description / context about the threat actor.
      - first_seen_date (int): estimated threat actor's first seen date of activity (UTC timestamp).
      - last_seen_date (int): estimated threat actor's last seen date of activity (UTC timestamp).
      - last_modification_date (int): last time when the threat actor was updated (UTC timestamp).
      - related_entities_count (int): estimated number of related IOCs to the threat actor.
      - source_region (str): threat actor's source region.
      - sponsor_region (str): region sponsoring the threat actor.
      - targeted_industries (list[str]): list of industries the threat actor has targeted.
      - targeted_regions (list[str]): list of regions the threat actor has targeted.
  
    Args:
      profile_id (str): Threat Profile identifier at Google Threat Intelligence.
  
    Returns:
      Threat Profile object.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object(
          client,
          "threat_profiles",
          "threat_profile",
          profile_id,
      )
    return utils.sanitize_response(res)

• gti_mcp/tools/threat_profiles.py:54-54 (registration)

  The @server.tool() decorator registers the get_threat_profile function as an MCP tool. The decorator comes from FastMCP and makes the function discoverable as an available tool in the MCP server.

  @server.tool()

• gti_mcp/utils.py:29-84 (helper)

  Helper function fetch_object that handles API requests to VirusTotal/Google Threat Intelligence. It fetches objects by resource type and ID, handles errors, and returns the response as a dictionary with the id included.

  async def fetch_object(
      vt_client: vt.Client,
      resource_collection_type: str,
      resource_type: str,
      resource_id: str,
      attributes: list[str] | None = None,
      relationships: list[str] | None = None,
      params: dict[str, typing.Any] | None = None):
    """Fetches objects from Google Threat Intelligence API."""
    logging.info(
        f"Fetching comprehensive {resource_collection_type} "
        f"report for id: {resource_id}")
    
    params = {k: v for k, v in params.items()} if params else {}
  
    # Retrieve a selection of object attributes and/or relationships.
    if attributes:
      params["attributes"] = ",".join(attributes)
    if relationships:
      params["relationships"] = ",".join(relationships)
  
    try:
      obj = await vt_client.get_object_async(
          f"/{resource_collection_type}/{resource_id}", params=params)
  
      if obj.error:
        logging.error(
            f"Error fetching main {resource_type} report for {resource_id}: {obj.error}"
        )
        return {
            "error": f"Failed to get main {resource_type} report: {obj.error}",
            # "details": report.get("details"),
        }
    except vt.error.APIError as e:
      logging.warning(
          f"VirusTotal API Error fetching {resource_type} {resource_id}: {e.code} - {e.message}"
      )
      return {
          "error": f"VirusTotal API Error: {e.code} - {e.message}",
          "details": f"The requested {resource_type} '{resource_id}' could not be found or there was an issue with the API request."
      }
    except Exception as e:
      logging.exception(
          f"Unexpected error fetching {resource_type} {resource_id}: {e}"
      )
      return {"error": "An unexpected internal error occurred."}
  
    # Build response.
    obj_dict = obj.to_dict()
    obj_dict['id'] = obj.id
    if 'aggregations' in obj_dict['attributes']:
      del obj_dict['attributes']['aggregations']
  
    logging.info(
        f"Successfully generated concise threat summary for id: {resource_id}")
    return obj_dict

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

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

  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