diff --git a/cloudinary/api.py b/cloudinary/api.py
index b905e45b..eb6511c7 100644
--- a/cloudinary/api.py
+++ b/cloudinary/api.py
@@ -1323,6 +1323,189 @@ def update_streaming_profile(name, **options):
     return call_json_api('PUT', uri, params, **options)
 
 
+def triggers(**options):
+    """
+    Lists all notification triggers.
+
+    :param options: Additional options.
+    :keyword str event_type: Restricts the list to triggers of a single event type.
+    :return: A dictionary with a "triggers" key and a "total" count.
+    :rtype: Response
+    """
+    params = {"event_type": options.pop("event_type", None)}
+    return call_json_api("get", ["triggers"], params, **options)
+
+
+def create_trigger(uri, event_type, uri_type=None, **options):
+    """
+    Creates a notification trigger.
+
+    :param uri: The destination of the notification. An https:// webhook URL, or a
+                poll:// destination when uri_type is "poll" - either "poll://<channel>"
+                for a named channel or "poll://*" for an anonymous one addressed by batch_id.
+    :type uri: str
+    :param event_type: The event that fires the trigger. One of:
+                       "all", "access_control_changed", "bulk_refresh_auto_fetch",
+                       "create_folder", "delete", "delete_by_token", "delete_folder",
+                       "eager", "error", "explode", "generate_archive", "info",
+                       "invalidate_custom_cdn", "moderation", "moderation_summary",
+                       "move", "move_or_rename_asset_folder", "multi", "publish",
+                       "rename", "report", "resource_context_changed",
+                       "resource_display_name_changed", "resource_metadata_changed",
+                       "resource_tags_changed", "restore_asset_version", "sprite",
+                       "upload".
+                       The server owns this list and may extend it; values are passed
+                       through unvalidated.
+    :type event_type: str
+    :param uri_type: How the notification is delivered:
+                     - "webhook" (the default): delivers the notification as an HTTP POST
+                       to the destination URL.
+                     - "flow": dispatches the notification to a Cloudinary flow.
+                     - "poll": buffers the notification for retrieval with `notifications`
+                       instead of an outbound HTTP request.
+    :type uri_type: str, optional
+    :param options: Additional options.
+    :return: The created trigger.
+    :rtype: Response
+    """
+    params = {"uri": uri, "event_type": event_type, "uri_type": uri_type}
+    return call_json_api("post", ["triggers"], params, **options)
+
+
+def update_trigger(trigger_id, new_uri, **options):
+    """
+    Updates the destination URI of an existing trigger.
+
+    :param trigger_id: The ID of the trigger to update.
+    :type trigger_id: str
+    :param new_uri: The new destination URI.
+    :type new_uri: str
+    :param options: Additional options.
+    :return: A dictionary with a "message" key.
+    :rtype: Response
+    """
+    return call_json_api("put", ["triggers", trigger_id], {"new_uri": new_uri}, **options)
+
+
+def delete_trigger(trigger_id, **options):
+    """
+    Deletes a notification trigger.
+
+    :param trigger_id: The ID of the trigger to delete.
+    :type trigger_id: str
+    :param options: Additional options.
+    :return: A dictionary with a "message" key.
+    :rtype: Response
+    """
+    return call_json_api("delete", ["triggers", trigger_id], {}, **options)
+
+
+def test_trigger(trigger_id, sample_data=None, **options):
+    """
+    Evaluates a trigger's filter against sample data, without delivering a notification.
+
+    :param trigger_id: The ID of the trigger to test.
+    :type trigger_id: str
+    :param sample_data: The sample notification payload to evaluate the filter against.
+    :type sample_data: dict, optional
+    :param options: Additional options.
+    :return: A dictionary with "trigger_id", "filter_present" and "filter_result" keys.
+    :rtype: Response
+    """
+    params = {"sample_data": sample_data}
+    return call_json_api("post", ["triggers", trigger_id, "test"], params, **options)
+
+
+def notifications(channel=None, batch_id=None, max_messages=None, wait_seconds=None,
+                  visibility_timeout=None, **options):
+    """
+    Gets notifications buffered for a poll destination.
+
+    Long-polls exactly one addressing dimension - a named channel or a batch_id. Holds the
+    connection up to wait_seconds, returning as soon as messages are claimed. A claimed
+    message stays invisible for visibility_timeout seconds and is then redelivered, unless
+    it is acknowledged with `ack_notifications` first.
+
+    Each message carries the parsed notification as "payload" and the byte-exact signed
+    bytes, base64url encoded, as "signed_payload". Verify signatures against
+    signed_payload, never against a re-serialization of payload::
+
+        raw = cloudinary.utils.base64url_decode(message["signed_payload"])
+        cloudinary.utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"])
+
+    :param channel: The named channel to drain ("poll://<channel>" destinations).
+                    Mutually exclusive with batch_id.
+    :type channel: str, optional
+    :param batch_id: The batch to drain ("poll://*" anonymous destinations).
+                     Mutually exclusive with channel.
+    :type batch_id: str, optional
+    :param max_messages: The maximum number of messages to return. 1-100, default 10.
+    :type max_messages: int, optional
+    :param wait_seconds: The maximum seconds to hold the long-poll open when empty.
+                         0-60, default 20.
+    :type wait_seconds: int, optional
+    :param visibility_timeout: The seconds a claimed message stays invisible before
+                               redelivery. 1-600, default 30.
+    :type visibility_timeout: int, optional
+    :param options: Additional options.
+    :keyword int timeout: The HTTP timeout, also configurable via `cloudinary.config(timeout=...)`.
+                          When set, it must exceed wait_seconds, or the long poll is cut off
+                          client-side before the server answers.
+    :return: Zero or more claimed messages, as {"messages": [...]}. The "messages" key is
+             absent when nothing is buffered - read it with .get("messages", []).
+    :rtype: Response
+    :raises ValueError: If neither or both of channel and batch_id are given.
+    """
+    if bool(channel) == bool(batch_id):
+        raise ValueError("Supply exactly one of 'channel' or 'batch_id'")
+
+    params = {
+        "channel": channel,
+        "batch_id": batch_id,
+        "max_messages": max_messages,
+        "wait_seconds": wait_seconds,
+        "visibility_timeout": visibility_timeout,
+    }
+
+    return __call_notifications_api("get", ["messages"], params, **options)
+
+
+def ack_notifications(receipt_handles, **options):
+    """
+    Acknowledges claimed notifications, deleting them from the buffer.
+
+    Unacknowledged notifications are redelivered once their visibility timeout expires.
+
+    :param receipt_handles: The receipt handles of the notifications to acknowledge, as
+                            returned by `notifications`. A single handle may be passed as a
+                            string. Maximum 100 per call.
+    :type receipt_handles: list[str] or str
+    :param options: Additional options.
+    :return: The result of the call.
+    :rtype: Response
+    """
+    if isinstance(receipt_handles, string_types):
+        receipt_handles = [receipt_handles]
+
+    return __call_notifications_api("post", ["messages", "ack"],
+                                    {"receipt_handles": receipt_handles}, **options)
+
+
+def __call_notifications_api(method, uri, params, **options):
+    """
+    Private function that assists with performing an API call to the notifications module.
+
+    :param method: The HTTP method. Valid methods: get, post, put, delete
+    :param uri: REST endpoint of the API (without 'notifications' or the cloud name)
+    :param params: Query/body parameters passed to the method
+    :param options: Additional options
+    :rtype: Response
+    :internal
+    """
+    return _call_v2_api(method, uri, params, module="notifications", **options)
+
+
 def only(source, *keys):
     """
     Returns a dictionary containing only the specified keys from the source.
@@ -1384,7 +1567,7 @@ def __delete_resource_params(options, **params):
     :internal
     """
     p = dict(transformations=utils.build_eager(options.get('transformations')),
-             **only(options, "keep_original", "next_cursor", "invalidate"))
+             **only(options, "keep_original", "next_cursor", "invalidate", "batch_id"))
     p.update(params)
     return p
 
diff --git a/cloudinary/api_client/call_api.py b/cloudinary/api_client/call_api.py
index 0e93b33c..66b465fd 100644
--- a/cloudinary/api_client/call_api.py
+++ b/cloudinary/api_client/call_api.py
@@ -43,15 +43,41 @@ def call_json_api(method, uri, params, **options):
     return _call_api(method, uri, params=params, body=data, headers={'Content-Type': 'application/json'}, **options)
 
 
-def _call_v2_api(method, uri, params, **options):
-    return call_json_api(method, uri, params=params, api_version='v2', **options)
+def _call_v2_api(method, uri, params, module=None, **options):
+    """Private function that assists with performing a v2 API call.
+
+    :param method: The HTTP method. Valid methods: get, post, put, delete
+    :param uri: REST endpoint of the API (without the module or the cloud name)
+    :param params: Query/body parameters passed to the method
+    :param module: The v2 module the endpoint belongs to, for module-first endpoints
+                   (`/v2/{module}/{cloud_name}/...`). Omit for context-first endpoints
+                   (`/v2/{cloud_name}/...`).
+    :param options: Additional options
+    :rtype: Response
+    """
+    return call_json_api(method, uri, params=params, api_version='v2', module=module, **options)
 
 
 def call_api(method, uri, params, **options):
     return _call_api(method, uri, params=params, **options)
 
 
-def _call_api(method, uri, params=None, body=None, headers=None, extra_headers=None, **options):
+def _call_api(method, uri, params=None, body=None, headers=None, extra_headers=None, module=None, **options):
+    """
+    Performs an API call.
+
+    :param method: The HTTP method. Valid methods: get, post, put, delete
+    :param uri: REST endpoint of the API, as a list of path segments
+    :param params: Query/body parameters passed to the method
+    :param body: An already serialized request body
+    :param headers: Request headers
+    :param extra_headers: Additional headers, merged into headers
+    :param module: The API module the endpoint belongs to. When given, the module precedes
+                   the cloud name in the URL (`/{api_version}/{module}/{cloud_name}/...`),
+                   which is the v2 API convention. When omitted, the cloud name comes first.
+    :param options: Additional options
+    :rtype: Response
+    """
     prefix = options.pop("upload_prefix",
                          cloudinary.config().upload_prefix) or "https://api.cloudinary.com"
     cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name)
@@ -66,7 +92,7 @@ def _call_api(method, uri, params=None, body=None, headers=None, extra_headers=N
     auth = {"key": api_key, "secret": api_secret, "oauth_token": oauth_token}
 
     api_version = options.pop("api_version", cloudinary.API_VERSION)
-    api_url = "/".join([prefix, api_version, cloud_name] + uri)
+    api_url = "/".join(filter(None, [prefix, api_version, module, cloud_name] + uri))
 
     if body is not None:
         options["body"] = body
diff --git a/cloudinary/api_client/execute_request.py b/cloudinary/api_client/execute_request.py
index 16d0bf45..ca299195 100644
--- a/cloudinary/api_client/execute_request.py
+++ b/cloudinary/api_client/execute_request.py
@@ -26,7 +26,8 @@
     409: AlreadyExists,
     420: RateLimited,
     429: RateLimited,
-    500: GeneralError
+    500: GeneralError,
+    503: GeneralError
 }
 
 
@@ -38,6 +39,7 @@ def __init__(self, result, response, **kwargs):
         self.rate_limit_allowed = safe_cast(response.headers.get("x-featureratelimit-limit"), int)
         self.rate_limit_reset_at = safe_cast(response.headers.get("x-featureratelimit-reset"), email.utils.parsedate)
         self.rate_limit_remaining = safe_cast(response.headers.get("x-featureratelimit-remaining"), int)
+        self.request_id = response.headers.get("x-request-id")
 
 
 def execute_request(http_connector, method, params, headers, auth, api_url, **options):
@@ -58,8 +60,9 @@ def execute_request(http_connector, method, params, headers, auth, api_url, **op
 
     api_url = smart_escape(unquote(api_url))
     kw = {}
-    if "timeout" in options:
-        kw["timeout"] = options["timeout"]
+    timeout = options.get("timeout", cloudinary.config().timeout)
+    if timeout is not None:
+        kw["timeout"] = timeout
     if "body" in options:
         kw["body"] = options["body"]
 
diff --git a/cloudinary/uploader.py b/cloudinary/uploader.py
index 50b86cc2..a626da4f 100644
--- a/cloudinary/uploader.py
+++ b/cloudinary/uploader.py
@@ -157,6 +157,8 @@ def upload(file, **options):
         If True, performs analysis for cinemagraph creation.
     :keyword bool accessibility_analysis:
         If True, performs accessibility (image alt text) analysis.
+    :keyword str batch_id:
+        The batch identifier used to retrieve notifications from a `poll://*` destination.
     :keyword int timestamp:
         A UNIX timestamp to sign the request. Defaults to now().
     :keyword dict or list transformation:
@@ -934,18 +936,27 @@ def call_api(action, params, http_headers=None, return_error=False, unsigned=Fal
     except socket.error as e:
         raise Error("Socket error: {0!r}".format(e))
 
+    request_id = response.headers.get("x-request-id")
+
     try:
         result = json.loads(response.data.decode('utf-8'))
     except Exception as e:
-        raise Error("Error parsing server response ({0}) - {1}. Got - {2}"
-                    .format(response.status, response.data, e))
+        message = "Error parsing server response ({0}) - {1}. Got - {2}".format(
+            response.status, response.data, e)
+        if request_id:
+            message += ". Request ID: {0}".format(request_id)
+        raise Error(message)
 
     if "error" in result:
         if return_error:
             result["error"]["http_code"] = response.status
+            result["error"]["request_id"] = request_id
             return result
 
         exception_class = EXCEPTION_CODES.get(response.status) or Error
         raise exception_class(result["error"]["message"])
 
+    if request_id:
+        result["request_id"] = request_id
+
     return result
diff --git a/cloudinary/utils.py b/cloudinary/utils.py
index be21fb79..c43dd034 100644
--- a/cloudinary/utils.py
+++ b/cloudinary/utils.py
@@ -122,6 +122,7 @@
     "cinemagraph_analysis",
     "accessibility_analysis",
     "auto_chaptering",
+    "batch_id",
 ]
 
 __SERIALIZED_UPLOAD_PARAMS = [
@@ -1092,6 +1093,7 @@ def archive_params(**options):
     params = {
         "allow_missing": options.get("allow_missing"),
         "async": options.get("async"),
+        "batch_id": options.get("batch_id"),
         "expires_at": options.get("expires_at"),
         "flatten_folders": options.get("flatten_folders"),
         "flatten_transformations": options.get("flatten_transformations"),
@@ -1234,6 +1236,7 @@ def build_multi_and_sprite_params(**options):
         "mode": options.get("mode"),
         "timestamp": now(),
         "async": options.get("async"),
+        "batch_id": options.get("batch_id"),
         "notification_url": options.get("notification_url"),
         "tag": tag,
         "urls": urls,
@@ -1555,6 +1558,17 @@ def base64url_encode(data):
     return to_string(base64.urlsafe_b64encode(to_bytes(data)))
 
 
+def base64url_decode(data):
+    """
+    Url safe version of urlsafe_b64decode that restores the `=` padding if it was stripped.
+
+    :param data: Base64 URL safe encoded string, padded or not
+
+    :return: Decoded string
+    """
+    return to_string(base64.urlsafe_b64decode(to_bytes(data + "=" * (-len(data) % 4))))
+
+
 def encode_unicode_url(url_str):
     """
     Quote and encode possible unicode url string (applicable for python2)
@@ -1645,7 +1659,9 @@ def verify_notification_signature(body, timestamp, signature, valid_for=7200, al
     Verifies the authenticity of a notification signature
 
     :param body: Json of the request's body
-    :param timestamp: Unix timestamp. Can be retrieved from the X-Cld-Timestamp header
+    :param timestamp: Unix timestamp. Can be retrieved from the X-Cld-Timestamp header, or from
+                      the `timestamp` of a notification returned by `api.notifications`. Both
+                      deliver it as a string, which is accepted here.
     :param signature: Actual signature. Can be retrieved from the X-Cld-Signature header
     :param valid_for: The desired time in seconds for considering the request valid
     :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
@@ -1656,7 +1672,7 @@ def verify_notification_signature(body, timestamp, signature, valid_for=7200, al
     if not cloudinary.config().api_secret:
         raise Exception('Api secret key is empty')
 
-    if timestamp < time.time() - valid_for:
+    if int(timestamp) < time.time() - valid_for:
         return False
 
     if not isinstance(body, str):
@@ -1667,6 +1683,31 @@ def verify_notification_signature(body, timestamp, signature, valid_for=7200, al
         algorithm or cloudinary.config().signature_algorithm)
 
 
+def verify_notification(message, valid_for=7200, algorithm=None):
+    """
+    Verifies the authenticity of a signed notification message.
+
+    :param message: Signed message, with `signed_payload`, `signature` and `timestamp` keys
+    :type message: dict
+    :param valid_for: The desired time in seconds for considering the message valid
+    :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
+                      By default, uses `cloudinary.config().signature_algorithm`
+
+    :return: Boolean result of the validation
+    :raises ValueError: If the message is missing signed_payload, signature or timestamp
+    """
+    missing = [key for key in ("signed_payload", "signature", "timestamp") if key not in message]
+    if missing:
+        raise ValueError("Message is missing required key(s): {}".format(", ".join(missing)))
+
+    return verify_notification_signature(
+        base64url_decode(message["signed_payload"]),
+        message["timestamp"],
+        message["signature"],
+        valid_for=valid_for,
+        algorithm=algorithm)
+
+
 def get_http_connector(conf, options):
     """
     Used to create http connector, depends on api_proxy and disable_tcp_keep_alive configuration parameters.
diff --git a/test/helper_test.py b/test/helper_test.py
index 616a771b..84fe3b83 100644
--- a/test/helper_test.py
+++ b/test/helper_test.py
@@ -55,6 +55,8 @@
 
 ON_SUCCESS_STR = 'current_asset.update({tags: ["autocaption"]});'
 
+MOCK_REQUEST_ID = "e529e88d138f4013655501c4711233d3"
+
 try:
     # urllib3 2.x support
     # noinspection PyProtectedMember
@@ -217,11 +219,12 @@ def http_response_mock(body="", headers=None, status=200):
 def api_response_mock(body='{"foo":"bar"}'):
     return http_response_mock(body, {"x-featureratelimit-limit": '0',
                                      "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT',
-                                     "x-featureratelimit-remaining": '0'})
+                                     "x-featureratelimit-remaining": '0',
+                                     "x-request-id": MOCK_REQUEST_ID})
 
 
 def uploader_response_mock():
-    return http_response_mock('{"foo":"bar"}')
+    return http_response_mock('{"foo":"bar"}', {"x-request-id": MOCK_REQUEST_ID})
 
 
 def populate_large_file(file_io, size, chunk_size=4096):
diff --git a/test/test_api.py b/test/test_api.py
index ed26c03e..ce79f662 100644
--- a/test/test_api.py
+++ b/test/test_api.py
@@ -139,8 +139,8 @@ def test_http_connector(self):
         self.assertIsInstance(http, ProxyManager)
 
     @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
-    def test_rate_limits(self):
-        """ should include details of the account's rate limits"""
+    def test_response_metadata(self):
+        """ should include the account's rate limits and the request id"""
         results = [
             api.ping(),
             api.root_folders(),
@@ -156,6 +156,9 @@ def test_rate_limits(self):
             self.assertIsNotNone(result.rate_limit_reset_at)
             self.assertGreater(result.rate_limit_remaining, 0)
 
+            self.assertIsNotNone(result.request_id)
+            self.assertNotIn("request_id", result)
+
     @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
     def test01_resource_types(self):
         """ should allow listing resource_types """
diff --git a/test/test_archive.py b/test/test_archive.py
index acc62dce..52fc7007 100644
--- a/test/test_archive.py
+++ b/test/test_archive.py
@@ -53,16 +53,19 @@ def test_optional_parameters(self, mocker):
         """should allow optional parameters"""
         mocker.return_value = MOCK_RESPONSE
         expires_at = int(time.time()+3600)
+        batch_id = "batch_{}".format(UNIQUE_TEST_ID)
         uploader.create_zip(
             tags=[TEST_TAG],
             expires_at=expires_at,
             allow_missing=True,
             skip_transformation_name=True,
+            batch_id=batch_id,
         )
         params = get_params(mocker)
         self.assertEqual(params['expires_at'], expires_at)
         self.assertTrue(params['allow_missing'])
         self.assertTrue(params['skip_transformation_name'])
+        self.assertEqual(params['batch_id'], batch_id)
 
     @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
     def test_archive_url(self):
diff --git a/test/test_notifications.py b/test/test_notifications.py
new file mode 100644
index 00000000..4300c624
--- /dev/null
+++ b/test/test_notifications.py
@@ -0,0 +1,321 @@
+import hashlib
+import json
+import time
+import unittest
+
+from urllib3 import disable_warnings
+
+import cloudinary
+from cloudinary import api, uploader, utils
+from test.helper_test import (
+    TEST_IMAGE, UNIQUE_TEST_ID, UNIQUE_TAG, get_uri, get_params, get_method, api_response_mock,
+    get_json_body, URLLIB3_REQUEST, patch, cleanup_test_resources_by_tag
+)
+
+MOCK_RESPONSE = api_response_mock()
+
+BATCH_ID = "batch_{}".format(UNIQUE_TEST_ID)
+CHANNEL = "channel_{}".format(UNIQUE_TEST_ID)
+RECEIPT_HANDLE = "receipt_handle_{}".format(UNIQUE_TEST_ID)
+LIVE_CHANNEL = "live_{}".format(UNIQUE_TEST_ID)
+
+disable_warnings()
+
+
+def build_message(payload=None, timestamp=None, algorithm=hashlib.sha1, api_secret=None):
+    """
+    Builds a signed message in the shape the notifications service returns.
+
+    The signature is computed over the byte-exact payload that was signed, which the server
+    sends base64url encoded and unpadded in `signed_payload`.
+    """
+    if payload is None:
+        payload = {"notification_type": "upload", "public_id": "sample", "request_id": "req_1"}
+
+    if timestamp is None:
+        timestamp = int(time.time())
+
+    timestamp = str(timestamp)
+
+    if api_secret is None:
+        api_secret = cloudinary.config().api_secret
+
+    raw = json.dumps(payload, separators=(",", ":"))
+    signature = algorithm("{}{}{}".format(raw, timestamp, api_secret).encode("utf-8")).hexdigest()
+
+    return {
+        "receipt_handle": RECEIPT_HANDLE,
+        "batch_id": BATCH_ID,
+        "payload": payload,
+        "signed_payload": utils.base64url_encode(raw).rstrip("="),
+        "signature": signature,
+        "timestamp": timestamp,
+    }
+
+
+class NotificationsTest(unittest.TestCase):
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test01_notifications_by_batch_id(self, mocker):
+        """Should drain notifications addressed by batch_id from the module-first v2 endpoint"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.notifications(batch_id=BATCH_ID)
+
+        self.assertEqual(get_method(mocker), "GET")
+        self.assertTrue(get_uri(mocker).endswith("/v2/notifications/{}/messages".format(
+            cloudinary.config().cloud_name)))
+        self.assertEqual(get_params(mocker).get("batch_id"), BATCH_ID)
+        self.assertIsNone(get_params(mocker).get("channel"))
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test02_notifications_by_channel(self, mocker):
+        """Should drain notifications addressed by channel"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.notifications(channel=CHANNEL)
+
+        self.assertEqual(get_params(mocker).get("channel"), CHANNEL)
+        self.assertIsNone(get_params(mocker).get("batch_id"))
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test03_notifications_passes_poll_options(self, mocker):
+        """Should pass the poll tuning options through"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.notifications(batch_id=BATCH_ID, max_messages=25, wait_seconds=30,
+                          visibility_timeout=60)
+
+        params = get_params(mocker)
+        self.assertEqual(params.get("max_messages"), "25")
+        self.assertEqual(params.get("wait_seconds"), "30")
+        self.assertEqual(params.get("visibility_timeout"), "60")
+
+    def test04_notifications_requires_exactly_one_address(self):
+        """Should require exactly one of channel or batch_id, mirroring the server"""
+        with self.assertRaises(ValueError):
+            api.notifications()
+
+        with self.assertRaises(ValueError):
+            api.notifications(channel=CHANNEL, batch_id=BATCH_ID)
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test05_notifications_timeout_from_config(self, mocker):
+        """Should take the HTTP timeout from cloudinary.config() when not passed per call"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.notifications(batch_id=BATCH_ID)
+        self.assertNotIn("timeout", mocker.call_args[1])
+
+        cloudinary.config(timeout=90)
+        try:
+            api.notifications(batch_id=BATCH_ID)
+            self.assertEqual(mocker.call_args[1]["timeout"], 90)
+        finally:
+            cloudinary.config(timeout=None)
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test06_notifications_explicit_timeout_wins(self, mocker):
+        """Should pass a per-call timeout through untouched, over the configured one"""
+        mocker.return_value = MOCK_RESPONSE
+
+        cloudinary.config(timeout=90)
+        try:
+            api.notifications(batch_id=BATCH_ID, wait_seconds=60, timeout=1)
+            self.assertEqual(mocker.call_args[1]["timeout"], 1)
+        finally:
+            cloudinary.config(timeout=None)
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test07_ack_notifications(self, mocker):
+        """Should acknowledge a list of receipt handles"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.ack_notifications([RECEIPT_HANDLE, "other_handle"])
+
+        self.assertEqual(get_method(mocker), "POST")
+        self.assertTrue(get_uri(mocker).endswith("/v2/notifications/{}/messages/ack".format(
+            cloudinary.config().cloud_name)))
+        self.assertEqual(get_json_body(mocker),
+                         {"receipt_handles": [RECEIPT_HANDLE, "other_handle"]})
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test08_ack_notifications_single_handle(self, mocker):
+        """Should accept a single receipt handle as a string"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.ack_notifications(RECEIPT_HANDLE)
+
+        self.assertEqual(get_json_body(mocker), {"receipt_handles": [RECEIPT_HANDLE]})
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test09_verify_signature_of_polled_notification(self):
+        """Should verify a polled notification, with the timestamp as a string or an int
+
+        The service sends the timestamp as a string, as does the webhook X-Cld-Timestamp
+        header; both forms have to be accepted.
+        """
+        message = build_message()
+        raw = utils.base64url_decode(message["signed_payload"])
+
+        self.assertIsInstance(message["timestamp"], str)
+        self.assertTrue(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"]))
+        self.assertTrue(utils.verify_notification_signature(
+            raw, int(message["timestamp"]), message["signature"]))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test10_verify_signature_rejects_tampered_payload(self):
+        """Should reject a signed payload that was modified after signing"""
+        message = build_message()
+
+        raw = utils.base64url_decode(message["signed_payload"]).replace("sample", "evil!!")
+
+        self.assertFalse(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"]))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test11_verify_signature_rejects_reserialized_payload(self):
+        """Should not verify against a re-serialization of the parsed payload
+
+        Signatures cover the byte-exact signed payload, so key order, spacing and escaping all
+        have to be preserved - which is why signed_payload exists.
+        """
+        message = build_message()
+
+        reserialized = json.dumps(message["payload"])
+
+        self.assertNotEqual(reserialized, utils.base64url_decode(message["signed_payload"]))
+        self.assertFalse(utils.verify_notification_signature(
+            reserialized, message["timestamp"], message["signature"]))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test12_verify_signature_rejects_expired_message(self):
+        """Should reject a notification signed longer than valid_for ago"""
+        message = build_message(timestamp=int(time.time()) - 10000)
+        raw = utils.base64url_decode(message["signed_payload"])
+
+        self.assertFalse(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"]))
+        self.assertTrue(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"], valid_for=20000))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test13_verify_signature_with_sha256(self):
+        """Should verify a notification signed with the configured sha256 algorithm"""
+        message = build_message(algorithm=hashlib.sha256)
+        raw = utils.base64url_decode(message["signed_payload"])
+
+        self.assertTrue(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"], algorithm=utils.SIGNATURE_SHA256))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test14_verify_signature_wrong_secret(self):
+        """Should reject a notification signed with a different api_secret"""
+        message = build_message(api_secret="another_secret")
+        raw = utils.base64url_decode(message["signed_payload"])
+
+        self.assertFalse(utils.verify_notification_signature(
+            raw, message["timestamp"], message["signature"]))
+
+    def test15_base64url_decode_restores_stripped_padding(self):
+        """Should decode the unpadded base64url the server sends, for every payload length"""
+        for size in range(1, 6):
+            raw = json.dumps({"public_id": "a" * size})
+            unpadded = utils.base64url_encode(raw).rstrip("=")
+
+            self.assertEqual(utils.base64url_decode(unpadded), raw)
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test16_verify_notification_accepts_the_message(self):
+        """Should verify a message envelope directly, decoding signed_payload internally"""
+        self.assertTrue(utils.verify_notification(build_message()))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test17_verify_notification_matches_the_manual_two_step(self):
+        """Should agree with the verify(base64url_decode(...)) call it replaces
+
+        Pins the helper to the underlying verifier, rather than to a second implementation of
+        the same digest.
+        """
+        for message in (build_message(),
+                        build_message(api_secret="another_secret"),
+                        build_message(timestamp=int(time.time()) - 10000)):
+            expected = utils.verify_notification_signature(
+                utils.base64url_decode(message["signed_payload"]),
+                message["timestamp"], message["signature"])
+
+            self.assertEqual(utils.verify_notification(message), expected)
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test18_verify_notification_rejects_tampered_payload(self):
+        """Should reject a message whose signed_payload was modified after signing"""
+        message = build_message()
+        tampered = utils.base64url_decode(message["signed_payload"]).replace("sample", "evil!!")
+        message["signed_payload"] = utils.base64url_encode(tampered).rstrip("=")
+
+        self.assertFalse(utils.verify_notification(message))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test19_verify_notification_passes_through_options(self):
+        """Should honor valid_for and algorithm, so the envelope form is not less capable"""
+        expired = build_message(timestamp=int(time.time()) - 10000)
+
+        self.assertFalse(utils.verify_notification(expired))
+        self.assertTrue(utils.verify_notification(expired, valid_for=20000))
+
+        sha256 = build_message(algorithm=hashlib.sha256)
+
+        self.assertTrue(utils.verify_notification(sha256, algorithm=utils.SIGNATURE_SHA256))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test20_verify_notification_reports_a_missing_key(self):
+        """Should raise rather than return False when the envelope is incomplete
+
+        A malformed envelope is a programming error, not a failed signature - returning False
+        would report it as a forgery.
+        """
+        for key in ("signed_payload", "signature", "timestamp"):
+            message = build_message()
+            del message[key]
+
+            with self.assertRaises(ValueError) as raised:
+                utils.verify_notification(message)
+
+            self.assertIn(key, str(raised.exception))
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test21_poll_notification_end_to_end(self):
+        """Should deliver, verify and acknowledge a real upload notification"""
+        trigger = api.create_trigger("poll://" + LIVE_CHANNEL, "upload", uri_type="poll")
+
+        try:
+            uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG])
+
+            messages = api.notifications(channel=LIVE_CHANNEL,
+                                         wait_seconds=20)["messages"]
+
+            self.assertTrue(messages)
+            self.assertEqual(messages[0]["payload"]["notification_type"], "upload")
+            # For testing purposes only.
+            self.assertTrue(utils.verify_notification(messages[0]))
+
+            acked = api.ack_notifications([m["receipt_handle"] for m in messages])
+
+            self.assertEqual([r["status"] for r in acked["results"]],
+                             ["acked"] * len(messages))
+            self.assertEqual(api.notifications(channel=LIVE_CHANNEL,
+                                               wait_seconds=0).get("messages", []), [])
+        finally:
+            api.delete_trigger(trigger["id"])
+            cleanup_test_resources_by_tag([(UNIQUE_TAG,)])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_request_id.py b/test/test_request_id.py
new file mode 100644
index 00000000..bb9b7a77
--- /dev/null
+++ b/test/test_request_id.py
@@ -0,0 +1,98 @@
+import unittest
+
+from urllib3 import disable_warnings
+
+import cloudinary
+from cloudinary import api, uploader
+from cloudinary.exceptions import Error
+from test.helper_test import (
+    TEST_IMAGE, MOCK_REQUEST_ID, http_response_mock, URLLIB3_REQUEST, patch
+)
+
+REQUEST_ID = MOCK_REQUEST_ID
+
+disable_warnings()
+
+
+class RequestIdTest(unittest.TestCase):
+    """
+    Covers `request_id` propagation in cases a live server will not produce on demand: a
+    missing X-Request-Id header, a malformed body, and a returned rather than raised error.
+
+    The happy paths are covered live, in test_api.py's test_response_metadata for the Admin
+    API and test_uploader.py for the Upload API.
+    """
+
+    def setUp(self):
+        cloudinary.reset_config()
+        cloudinary.config(cloud_name="test123", api_key="1234", api_secret="b")
+
+    def tearDown(self):
+        cloudinary.reset_config()
+
+    @patch(URLLIB3_REQUEST)
+    def test_admin_response_without_request_id(self, mocker):
+        """Should leave request_id as None when the header is absent"""
+        mocker.return_value = http_response_mock('{"foo":"bar"}')
+
+        self.assertIsNone(api.ping().request_id)
+
+    @patch(URLLIB3_REQUEST)
+    def test_upload_response_without_request_id(self, mocker):
+        """Should not add a request_id key when the server did not report one"""
+        mocker.return_value = http_response_mock('{"public_id":"test"}')
+
+        result = uploader.upload(TEST_IMAGE)
+
+        self.assertNotIn("request_id", result)
+
+    @patch(URLLIB3_REQUEST)
+    def test_upload_request_id_is_always_the_header(self, mocker):
+        """Should report the X-Request-Id header, which is the transport level identifier
+
+        The Upload API returns no request_id of its own today; if it ever does, the header still
+        wins, so the key means one thing consistently.
+        """
+        mocker.return_value = http_response_mock('{"public_id":"test","request_id":"from_body"}',
+                                                 {"x-request-id": REQUEST_ID})
+
+        result = uploader.upload(TEST_IMAGE)
+
+        self.assertEqual(result["request_id"], REQUEST_ID)
+        self.assertIs(type(result), dict)
+
+    @patch(URLLIB3_REQUEST)
+    def test_upload_parse_error_prints_request_id(self, mocker):
+        """Should include the request id in an Upload API parsing error"""
+        mocker.return_value = http_response_mock("not json", {"x-request-id": REQUEST_ID},
+                                                 status=500)
+
+        with self.assertRaises(Error) as raised:
+            uploader.upload(TEST_IMAGE)
+
+        self.assertIn("Request ID: {}".format(REQUEST_ID), str(raised.exception))
+
+    @patch(URLLIB3_REQUEST)
+    def test_upload_parse_error_without_request_id_keeps_original_message(self, mocker):
+        """Should not add a request id suffix when the response header is absent"""
+        mocker.return_value = http_response_mock("not json", status=500)
+
+        with self.assertRaises(Error) as raised:
+            uploader.upload(TEST_IMAGE)
+
+        self.assertNotIn("Request ID:", str(raised.exception))
+
+    @patch(URLLIB3_REQUEST)
+    def test_upload_returned_error_carries_request_id(self, mocker):
+        """Should include the request id when the error is returned rather than raised"""
+        mocker.return_value = http_response_mock('{"error":{"message":"bad request"}}',
+                                                 {"x-request-id": REQUEST_ID}, status=400)
+
+        result = uploader.upload(TEST_IMAGE, return_error=True)
+
+        self.assertEqual(result["error"]["http_code"], 400)
+        self.assertEqual(result["error"]["request_id"], REQUEST_ID)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_triggers.py b/test/test_triggers.py
new file mode 100644
index 00000000..02b00d24
--- /dev/null
+++ b/test/test_triggers.py
@@ -0,0 +1,162 @@
+import unittest
+
+from urllib3 import disable_warnings
+
+import cloudinary
+from cloudinary import api
+from test.helper_test import (
+    UNIQUE_TEST_ID, get_uri, get_params, get_method, api_response_mock, get_json_body,
+    URLLIB3_REQUEST, patch
+)
+
+MOCK_RESPONSE = api_response_mock()
+
+TRIGGER_ID = "trigger_id_{}".format(UNIQUE_TEST_ID)
+WEBHOOK_URI = "https://example.com/notifications/{}".format(UNIQUE_TEST_ID)
+POLL_CHANNEL_URI = "poll://orders/{}".format(UNIQUE_TEST_ID)
+POLL_ANONYMOUS_URI = "poll://*"
+LIVE_POLL_URI = "poll://live_{}".format(UNIQUE_TEST_ID)
+
+disable_warnings()
+
+
+class TriggersTest(unittest.TestCase):
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test01_list_triggers(self, mocker):
+        """Should list all notification triggers"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.triggers()
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers"))
+        self.assertEqual(get_method(mocker), "GET")
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test02_list_triggers_by_event_type(self, mocker):
+        """Should list notification triggers of a single event type"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.triggers(event_type="upload")
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers"))
+        self.assertEqual(get_method(mocker), "GET")
+        self.assertEqual(get_params(mocker).get("event_type"), "upload")
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test03_create_webhook_trigger(self, mocker):
+        """Should create a webhook trigger"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.create_trigger(WEBHOOK_URI, "upload")
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers"))
+        self.assertEqual(get_method(mocker), "POST")
+        self.assertEqual(get_json_body(mocker),
+                         {"uri": WEBHOOK_URI, "event_type": "upload", "uri_type": None})
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test04_create_poll_trigger(self, mocker):
+        """Should create an anonymous poll trigger"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.create_trigger(POLL_ANONYMOUS_URI, "upload", uri_type="poll")
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers"))
+        self.assertEqual(get_method(mocker), "POST")
+        self.assertEqual(get_json_body(mocker), {
+            "uri": POLL_ANONYMOUS_URI,
+            "event_type": "upload",
+            "uri_type": "poll",
+        })
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test05_create_poll_channel_trigger(self, mocker):
+        """Should create a named channel poll trigger, passing the poll:// URI through"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.create_trigger(POLL_CHANNEL_URI, "all", uri_type="poll")
+
+        self.assertEqual(get_json_body(mocker), {
+            "uri": POLL_CHANNEL_URI,
+            "event_type": "all",
+            "uri_type": "poll",
+        })
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test06_update_trigger(self, mocker):
+        """Should update a trigger's destination URI"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.update_trigger(TRIGGER_ID, WEBHOOK_URI)
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers/{}".format(TRIGGER_ID)))
+        self.assertEqual(get_method(mocker), "PUT")
+        self.assertEqual(get_json_body(mocker), {"new_uri": WEBHOOK_URI})
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test08_delete_trigger(self, mocker):
+        """Should delete a trigger"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.delete_trigger(TRIGGER_ID)
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers/{}".format(TRIGGER_ID)))
+        self.assertEqual(get_method(mocker), "DELETE")
+        self.assertEqual(get_json_body(mocker), {})
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test09_test_trigger(self, mocker):
+        """Should evaluate a trigger's filter against sample data"""
+        mocker.return_value = MOCK_RESPONSE
+
+        sample_data = {"notification_type": "upload", "public_id": "sample"}
+        api.test_trigger(TRIGGER_ID, sample_data)
+
+        self.assertTrue(get_uri(mocker).endswith("/triggers/{}/test".format(TRIGGER_ID)))
+        self.assertEqual(get_method(mocker), "POST")
+        self.assertEqual(get_json_body(mocker), {"sample_data": sample_data})
+
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test10_trigger_lifecycle(self):
+        """Should create, list, update and delete a trigger against the live API"""
+        created = api.create_trigger(LIVE_POLL_URI, "upload", uri_type="poll")
+        trigger_id = created["id"]
+
+        try:
+            self.assertEqual(created["uri"], LIVE_POLL_URI)
+            self.assertEqual(created["uri_type"], "poll")
+            self.assertEqual(created["event_type"], "upload")
+
+            listed = api.triggers()
+            self.assertIn(trigger_id, [t["id"] for t in listed["triggers"]])
+
+            api.update_trigger(trigger_id, LIVE_POLL_URI + "_updated")
+
+            updated = [t for t in api.triggers()["triggers"] if t["id"] == trigger_id]
+            self.assertEqual(updated[0]["uri"], LIVE_POLL_URI + "_updated")
+        finally:
+            api.delete_trigger(trigger_id)
+
+        self.assertNotIn(trigger_id, [t["id"] for t in api.triggers()["triggers"]])
+
+    @patch(URLLIB3_REQUEST)
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test10_test_trigger_without_sample_data(self, mocker):
+        """Should send a null sample_data when it is not given; the server treats it as absent"""
+        mocker.return_value = MOCK_RESPONSE
+
+        api.test_trigger(TRIGGER_ID)
+
+        self.assertEqual(get_json_body(mocker), {"sample_data": None})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_uploader.py b/test/test_uploader.py
index 6a7cc608..3e60d722 100644
--- a/test/test_uploader.py
+++ b/test/test_uploader.py
@@ -159,6 +159,7 @@ def test_upload(self):
         result = uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG])
         self.assertEqual(result["width"], TEST_IMAGE_WIDTH)
         self.assertEqual(result["height"], TEST_IMAGE_HEIGHT)
+        self.assertIsNotNone(result["request_id"])
         expected_signature = utils.api_sign_request(
             dict(public_id=result["public_id"], version=result["version"]),
             cloudinary.config().api_secret)
@@ -486,7 +487,7 @@ def test_update_metadata(self):
 
         result = uploader.update_metadata(METADATA_FIELDS, public_ids)
 
-        self.assertEqual(result, {
+        self.assertObjectContainsSubset(result, {
             "public_ids": public_ids,
         })
 
@@ -1167,6 +1168,7 @@ def test_various_upload_parameters(self, request_mock):
             'regions': {"box_1": [[1, 2], [3, 4]], "box_2": [[5, 6], [7, 8]]},
             'auto_transcription': True,
             'auto_chaptering': True,
+            'batch_id': 'batch_{}'.format(UNIQUE_ID),
         }
 
         uploader.upload(TEST_IMAGE, **options)