Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/threat_profiles.py:147-173 (handler)

  The main handler and registration for get_threat_profile_associations_timeline tool. This async function takes profile_id, ctx, limit, and optional api_key parameters, and retrieves timeline associations from the VirusTotal API endpoint /threat_profiles/{profile_id}/timeline/associations. It uses the @server.tool() decorator for registration.

  @server.tool()
  async def get_threat_profile_associations_timeline(
      profile_id: str, ctx: Context, limit: int = 10, api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """Retrieves the associations timeline for the given Threat Profile.
  
    Some important response attributes:
      - event_type (str): the type of the timeline association such as Alias, Motivation,
                          Malware, Actor, Toolkit, Report, Campaign, etc.
      - event_entity (str): The name or value of the timeline association.
      - first_seen (int): Unix epoch UTC time (seconds) when the association
                          between the object and the threat profile was made.
      - last_seen (int): Unix epoch UTC time (seconds) of most recent observed
                          relationship between the object and the threat profile.
      - name (str): name of the object directly associated with the threat profile.
      - link (str): URL of the object directly associated with the threat profile
  
    Returns:
      List of dictionaries containing timeline associations.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.consume_vt_iterator(
          client,
          f"/threat_profiles/{profile_id}/timeline/associations",
          limit=limit,
      )
    return utils.sanitize_response([o.to_dict() for o in res])

• gti_mcp/utils.py:20-26 (helper)

  Helper function consume_vt_iterator that consumes a VirusTotal API iterator and returns a list of objects. Used by the handler to fetch timeline associations from the API.

  async def consume_vt_iterator(
      vt_client: vt.Client, endpoint: str, params: dict | None = None, limit: int = 10):
    """Consumes a vt.Iterator iterator and return the list of objects."""
    res = []
    async for obj in vt_client.iterator(endpoint, params=params, limit=limit):
      res.append(obj)
    return res

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

  Helper function sanitize_response that recursively removes empty dictionaries and lists from API responses. Used to clean up the timeline associations response before returning.

  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