Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/files.py:88-107 (handler)

  The main handler function for get_file_report tool. It takes a file hash (MD5/SHA-1/SHA-256), creates a VirusTotal client context, fetches the file report with specific relationships using the fetch_object helper, and returns a sanitized response with the file analysis data.

  @server.tool()
  async def get_file_report(hash: str, ctx: Context, api_key: str = None) -> typing.Dict[str, typing.Any]:
    """Get a comprehensive file analysis report using its hash (MD5/SHA-1/SHA-256).
  
    Returns a concise summary of key threat details including
    detection stats, threat classification, and important indicators.
    Parameters:
      hash (required): The MD5, SHA-1, or SHA-256 hash of the file to analyze.
    Example: '8ab2cf...', 'e4d909c290d0...', etc.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object(
          client,
          "files",
          "file",
          hash,
          relationships=FILE_KEY_RELATIONSHIPS,
          params={"exclude_attributes": "last_analysis_results"}
      )
    return utils.sanitize_response(res)

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

  The fetch_object helper function used by get_file_report to retrieve data from the Google Threat Intelligence API. It handles API errors, builds the request with specified attributes and relationships, and processes the response by removing aggregations from attributes.

  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)

  The sanitize_response helper function that recursively removes empty dictionaries, empty lists, and empty strings from API responses to clean up the output returned to the user.

  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/tools/files.py:76-85 (schema)

  The FILE_KEY_RELATIONSHIPS constant defines the list of relationships to fetch when calling get_file_report. These relationships include contacted domains, IPs, URLs, dropped files, embedded entities, and associations that provide comprehensive file analysis context.

  FILE_KEY_RELATIONSHIPS = [
      "contacted_domains",
      "contacted_ips",
      "contacted_urls",
      "dropped_files",
      "embedded_domains",
      "embedded_ips",
      "embedded_urls",
      "associations",
  ]

• gti_mcp/tools/files.py:88-88 (registration)

  The @server.tool() decorator registers the get_file_report function as an MCP tool with the FastMCP server, making it available for client invocation.

  @server.tool()