Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:384-406 (handler)

  The main handler function get_collection_timeline_events that retrieves timeline events from a collection via VirusTotal API. Decorated with @server.tool() for MCP registration.

  @server.tool()
  async def get_collection_timeline_events(id: str, ctx: Context, api_key: str = None):
    """Retrieves timeline events from the given collection, when available.
  
    This is super valuable curated information produced by security analysits at Google Threat Intelligence.
  
    We should fetch this information for campaigns and threat actors always.
  
    It's common to display the events grouped by the "event_category" field.
  
    Args:
      id (required): Collection identifier
    Return:
      List of events related to the given collection.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      resp = await client.get_async(f"/collections/{id}/timeline/events")
      if resp.status != 200:
          error_json = await resp.json_async()
          error_info = error_json.get("error", {})
          return [{"error": f"API Error: {error_info.get('message', 'Unknown error')}"}]
      data = await resp.json_async()
    return utils.sanitize_response(data.get("data", []))

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

  The @server.tool() decorator that registers get_collection_timeline_events as an MCP tool. The server instance is a FastMCP object from gti_mcp/server.py.

  @server.tool()

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

  Helper function sanitize_response that recursively removes empty dictionaries and lists from API responses, used by get_collection_timeline_events to clean up returned data.

  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

• tests/test_collections_errors.py:20-42 (helper)

  Test case for get_collection_timeline_events that verifies error handling when the API returns a non-200 status code (e.g., 404 for non-existent collection).

  async def test_get_collection_timeline_events_api_error_handled():
      """Test get_collection_timeline_events returns error dict on API error (e.g. 404)."""
      mock_ctx = MagicMock(spec=Context)
  
      mock_response = MagicMock()
      mock_response.status = 404
      async def json_async():
          return {"error": {"message": "Collection not found"}}
      mock_response.json_async = json_async
  
      mock_client_instance = MagicMock()
      mock_client_instance.get_async = AsyncMock(return_value=mock_response)
  
      mock_vt_client = MagicMock()
      mock_vt_client.__aenter__ = AsyncMock(return_value=mock_client_instance)
      mock_vt_client.__aexit__ = AsyncMock(return_value=None)
  
      with patch("gti_mcp.tools.collections.vt_client", return_value=mock_vt_client):
          result = await collections.get_collection_timeline_events(id="non_existent_id", ctx=mock_ctx)
          assert isinstance(result, list)
          assert len(result) == 1
          assert "error" in result[0]
          assert "Collection not found" in result[0]["error"]