get_collection_report

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:57-83 (handler)

  The main handler for the 'get_collection_report' tool. It uses the @server.tool() decorator to register as an MCP tool, accepts an id, ctx, and optional api_key, and calls utils.fetch_object to retrieve a collection from the Google Threat Intelligence API.

  @server.tool()
  async def get_collection_report(id: str, ctx: Context, api_key: str = None) -> typing.Dict[str, typing.Any]:
    """At Google Threat Intelligence, threats are modeled as "collections". This tool retrieves them from the platform.
  
    They have different collections types like: 
      - "malware-family"
      - "threat-actor"
      - "campaign"
      - "report"
      - "collection". 
  
    You can find the collection type in the "collection_type" field.
  
    Args:
      id (required): Google Threat Intelligence identifier.
    Returns:
      A collection object. Put attention to the collection type to correctly understand what it represents.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.fetch_object(
          client,
          "collections",
          "collection",
          id,
          relationships=COLLECTION_KEY_RELATIONSHIPS,
          params={"exclude_attributes": COLLECTION_EXCLUDED_ATTRS})
    return res

• gti_mcp/tools/collections.py:57-57 (registration)

  The tool is registered via the @server.tool() decorator on the get_collection_report function, where 'server' is a FastMCP instance from gti_mcp/server.py.

  @server.tool()

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

  The fetch_object helper function called by get_collection_report. It fetches data from VT API using client.get_object_async with the given collection type ('collections'), id, relationships, and params.

  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/tools/collections.py:41-44 (schema)

  COLLECTION_KEY_RELATIONSHIPS and COLLECTION_EXCLUDED_ATTRS: constants that define the schema/configuration used by get_collection_report for fetching related data.

  COLLECTION_KEY_RELATIONSHIPS = [
      "associations",
  ]
  COLLECTION_EXCLUDED_ATTRS = ",".join(["aggregations"])

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

  The collections module (which contains get_collection_report) is imported into the tools package, which is loaded by server.py line 73: 'from gti_mcp.tools import *'.

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