Azure · Eldert Grootenboer (EldertGrootenboer) · Sep 2, 2026 · Sep 2, 2026 · Sep 2, 2026 · Sep 3, 2026
@@ -14,6 +14,7 @@
 
 ### Features Added
 
+- Added sync and async `ServiceBusReceiver.delete_messages()` and `purge_messages()` methods. Basic and Standard support up to 500 messages per request, Premium supports up to 4,000, and purge handles smaller batches caused by large messages.
 - Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979))
 - Added `ServiceBusClient.list_queue_sessions()` and `ServiceBusClient.list_subscription_sessions()` (sync and async) to list session IDs for entities with active messages or stored session state, with optional filtering by session-state update timestamp. The methods return an `ItemPaged[str]` (`AsyncItemPaged[str]` on the async client) so callers can iterate every session transparently or page with `by_page()`. Implements the `com.microsoft:get-message-sessions` management operation. ([#46575](https://github.com/Azure/azure-sdk-for-python/pull/46575))
 - Added `sql_filter_count` and `correlation_filter_count` properties to `TopicRuntimeProperties`, exposing the total number of SQL filters and correlation filters across all of a topic's subscriptions.

@@ -1,3 +1,3 @@
-apiMdSha256: 70b3fbd3239df511d17fde454a717f9ef0a11f125b096da26d9c815ab0120a17
+apiMdSha256: 87c7a74eca4456c54fc09ea826778182be53acc571095c11a8ebe7b12a981dd0
 parserVersion: 0.3.31
 pythonVersion: 3.14.0
@@ -11,6 +11,7 @@
 from ._servicebus_sender import ServiceBusSender
 from ._servicebus_receiver import ServiceBusReceiver
 from ._servicebus_session import ServiceBusSession
+from ._models import DeleteMessagesResult, PurgeMessagesResult
 from ._common.message import (
     ServiceBusMessage,
     ServiceBusMessageBatch,
@@ -43,6 +44,8 @@
     "ServiceBusReceiver",
     "ServiceBusSession",
     "ServiceBusSender",
+    "DeleteMessagesResult",
+    "PurgeMessagesResult",
     "TransportType",
     "AutoLockRenewer",
     "parse_connection_string",

@@ -38,6 +38,7 @@
     MGMT_REQUEST_OP_TYPE_ENTITY_MGMT,
     ASSOCIATEDLINKPROPERTYNAME,
     REQUEST_RESPONSE_TIMEOUT,
+    NEXT_AVAILABLE_SESSION,
 )
 
 if TYPE_CHECKING:
@@ -381,6 +382,9 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
     ) -> Any:
         require_last_exception = kwargs.pop("require_last_exception", False)
         operation_requires_timeout = kwargs.pop("operation_requires_timeout", False)
+        suppress_next_session_timeout_message = kwargs.pop(
+            "suppress_next_session_timeout_message", False
+        )
         retried_times = 0
         max_retries = self._config.retry_total
 
@@ -408,7 +412,10 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
                         self._container_id,
                         last_exception,
                     )
-                    if isinstance(last_exception, OperationTimeoutError):
+                    if isinstance(last_exception, OperationTimeoutError) and (
+                        getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION
+                        and not suppress_next_session_timeout_message
+                    ):
                         description = (
                             "If trying to receive from NEXT_AVAILABLE_SESSION, "
                             "use max_wait_time on the ServiceBusReceiver to control the"
@@ -423,6 +430,9 @@ def _do_retryable_operation(  # pylint: disable=inconsistent-return-statements
                     retried_times=retried_times,
                     last_exception=last_exception,
                     abs_timeout_time=abs_timeout_time,
+                    suppress_next_session_timeout_message=(
+                        suppress_next_session_timeout_message
+                    ),
                 )
 
     def _backoff(
@@ -431,6 +441,7 @@ def _backoff(
         last_exception: Exception,
         abs_timeout_time: Optional[float] = None,
         entity_name: Optional[str] = None,
+        suppress_next_session_timeout_message: bool = False,
     ) -> None:
         entity_name = entity_name or self._container_id
         backoff = _get_backoff_time(
@@ -454,7 +465,10 @@ def _backoff(
                 entity_name,
                 last_exception,
             )
-            if isinstance(last_exception, OperationTimeoutError):
+            if isinstance(last_exception, OperationTimeoutError) and (
+                getattr(self, "_session_id", None) == NEXT_AVAILABLE_SESSION
+                and not suppress_next_session_timeout_message
+            ):
                 description = (
                     "If trying to receive from NEXT_AVAILABLE_SESSION, "
                     "use max_wait_time on the ServiceBusReceiver to control the"
@@ -553,8 +567,63 @@ def _mgmt_request_response_with_retry(
     def _open(self):
         raise ValueError("Subclass should override the method.")
 
-    def _open_with_retry(self):
-        return self._do_retryable_operation(self._open)
+    def _open_with_timeout(self, timeout: float):
+        del timeout
+        return self._open()
+
+    def _open_with_retry(
+        self,
+        timeout: Optional[float] = None,
+        *,
+        suppress_next_session_timeout_message: bool = False,
+    ):
+        def open_with_timeout(timeout: Optional[float] = None):
+            if timeout is not None and timeout <= 0:
+                raise OperationTimeoutError()
+            if timeout is None:
+                return self._open()
+            return self._open_with_timeout(timeout)
+
+        return self._do_retryable_operation(
+            open_with_timeout,
+            timeout=timeout,
+            operation_requires_timeout=timeout is not None,
+            suppress_next_session_timeout_message=(
+                suppress_next_session_timeout_message
+            ),
+        )
+
+    def _open_mgmt_link_with_retry(
+        self,
+        timeout: Optional[float] = None,
+        *,
+        suppress_next_session_timeout_message: bool = False,
+    ):
+        def open_mgmt_link(timeout: Optional[float] = None):
+            if timeout is not None and timeout <= 0:
+                raise OperationTimeoutError()
+            start_time = time.monotonic()
+            if timeout is None:
+                self._open()
+            else:
+                self._open_with_timeout(timeout)
+                timeout -= time.monotonic() - start_time
+                if timeout <= 0:
+                    raise OperationTimeoutError()
+            return self._amqp_transport.mgmt_client_setup(
+                self._handler,
+                node=self._mgmt_target.encode(self._config.encoding),
+                timeout=timeout,
+            )
+
+        return self._do_retryable_operation(
+            open_mgmt_link,
+            timeout=timeout,
+            operation_requires_timeout=timeout is not None,
+            suppress_next_session_timeout_message=(
+                suppress_next_session_timeout_message
+            ),
+        )
 
     def _close_handler(self):
         if self._handler:

@@ -52,6 +52,7 @@
 REQUEST_RESPONSE_ADD_RULE_OPERATION = VENDOR + b":add-rule"
 REQUEST_RESPONSE_REMOVE_RULE_OPERATION = VENDOR + b":remove-rule"
 REQUEST_RESPONSE_GET_RULES_OPERATION = VENDOR + b":enumerate-rules"
+REQUEST_RESPONSE_BATCH_DELETE_MESSAGES_OPERATION = VENDOR + b":batch-delete-messages"
 
 CONTAINER_PREFIX = "servicebus.pysdk-"
 JWT_TOKEN_SCOPE = "https://servicebus.azure.net//.default"
@@ -82,6 +83,8 @@
 MGMT_REQUEST_RECEIVER_SETTLE_MODE = "receiver-settle-mode"
 MGMT_REQUEST_FROM_SEQUENCE_NUMBER = "from-sequence-number"
 MGMT_REQUEST_MAX_MESSAGE_COUNT = "message-count"
+MGMT_REQUEST_MESSAGE_COUNT = "message-count"
+MGMT_REQUEST_ENQUEUED_TIME_UTC = "enqueued-time-utc"
 MGMT_REQUEST_MESSAGE = "message"
 MGMT_REQUEST_MESSAGES = "messages"
 MGMT_REQUEST_MESSAGE_ID = "message-id"

@@ -69,6 +69,25 @@ def peek_op(  # pylint: disable=inconsistent-return-statements
     )
 
 
+def batch_delete_op(  # pylint: disable=inconsistent-return-statements
+    status_code, message, description, amqp_transport, max_message_count
+):
+    condition = message.application_properties.get(MGMT_RESPONSE_MESSAGE_ERROR_CONDITION)
+    if status_code == 200 or (status_code == 404 and condition == ERROR_CODE_MESSAGE_NOT_FOUND):
+        deleted_count = message.value.get(b"message-count") if isinstance(message.value, dict) else None
+        if (
+            isinstance(deleted_count, bool)
+            or not isinstance(deleted_count, int)
+            or deleted_count < 0
+            or deleted_count > max_message_count
+        ):
+            raise ValueError("Batch delete response did not contain a valid message-count.")
+        return deleted_count
+    amqp_transport.handle_amqp_mgmt_error(
+        _LOGGER, "Batch delete messages failed.", condition, description, status_code
+    )
+
+
 def list_sessions_op(  # pylint: disable=inconsistent-return-statements
     status_code, message, description, amqp_transport
 ):

@@ -72,6 +72,15 @@
 # Number of seconds between the Unix epoch (1/1/1970) and year 1 CE.
 # This is the lowest value that can be represented by an AMQP timestamp.
 CE_ZERO_SECONDS: int = -62_135_596_800
+EPOCH_UTC: datetime.datetime = datetime.datetime(1970, 1, 1, tzinfo=TZ_UTC)
+
+
+def datetime_to_timestamp_ms(value: datetime.datetime) -> int:
+    if value.tzinfo is None:
+        normalized = value.replace(tzinfo=TZ_UTC)
+    else:
+        normalized = value.astimezone(TZ_UTC)
+    return (normalized - EPOCH_UTC) // datetime.timedelta(milliseconds=1)
 
 def utc_from_timestamp(timestamp: float) -> datetime.datetime:
     """

@@ -0,0 +1,30 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+
+
+class DeleteMessagesResult:
+    """The result of a batch delete operation.
+
+    :ivar int deleted_message_count: The number of messages deleted by the service.
+    """
+
+    def __init__(self, deleted_message_count: int) -> None:
+        self._deleted_message_count = deleted_message_count
+
+    @property
+    def deleted_message_count(self) -> int:
+        """The number of messages deleted by the service.
+
+        :rtype: int
+        """
+        return self._deleted_message_count
+
+
+class PurgeMessagesResult(DeleteMessagesResult):
+    """The result of a purge operation.
+
+    :ivar int deleted_message_count: The total number of messages deleted by the service.
+    """
@@ -368,29 +368,50 @@ async def mgmt_request_async(
         :rtype: ~pyamqp.message.Message
         """
 
-        # The method also takes "status_code_field" and "status_description_field"
-        # keyword arguments as alternate names for the status code and description
-        # in the response body. Those two keyword arguments are used in Azure services only.
-        async with self._mgmt_link_lock_async:
-            try:
-                mgmt_link = self._mgmt_links[node]
-            except KeyError:
-                mgmt_link = ManagementOperation(self._session, endpoint=node, **kwargs)
-                self._mgmt_links[node] = mgmt_link
-                await mgmt_link.open()
-
-        while not await self.client_ready_async():
-            await asyncio.sleep(0.05)
-
-        while not await mgmt_link.ready():
-            await self._connection.listen(wait=False)
+        mgmt_link = await self.open_mgmt_link_async(node=node, timeout=timeout, **kwargs)
 
         operation_type = operation_type or b"empty"
         status, description, response = await mgmt_link.execute(
             message, operation=operation, operation_type=operation_type, timeout=timeout
         )
         return status, description, response
 
+    async def open_mgmt_link_async(self, node: str = "$management", timeout: float = 0, **kwargs):
+        """Open and wait for a management link without dispatching a request.
+
+        :param str node: Management target.
+        :param float timeout: Timeout in seconds.
+        :returns: The opened management link.
+        :rtype: ~pyamqp.aio.management_link_async.ManagementOperation
+        """
+        start_time = time.monotonic()
+        mgmt_link = None
+        try:
+            async with self._mgmt_link_lock_async:
+                try:
+                    mgmt_link = self._mgmt_links[node]
+                except KeyError:
+                    mgmt_link = ManagementOperation(self._session, endpoint=node, **kwargs)
+                    self._mgmt_links[node] = mgmt_link
+                    await mgmt_link.open()
+            while not await self.client_ready_async():
+                if timeout and time.monotonic() - start_time >= timeout:
+                    raise TimeoutError("Management link setup timed out.")
+                await asyncio.sleep(0.05)
+
+            while not await mgmt_link.ready():
+                if timeout and time.monotonic() - start_time >= timeout:
+                    raise TimeoutError("Management link setup timed out.")
+                await self._connection.listen(wait=False)
+            return mgmt_link
+        except BaseException:
+            if mgmt_link is not None:
+                async with self._mgmt_link_lock_async:
+                    if self._mgmt_links.get(node) is mgmt_link:
+                        self._mgmt_links.pop(node, None)
+                await mgmt_link.close()
+            raise
+
 
 class SendClientAsync(SendClientSync, AMQPClientAsync):
     """An asynchronous AMQP client.

@@ -455,28 +455,50 @@ def mgmt_request(
         :rtype: ~pyamqp.message.Message
         """
 
-        # The method also takes "status_code_field" and "status_description_field"
-        # keyword arguments as alternate names for the status code and description
-        # in the response body. Those two keyword arguments are used in Azure services only.
-        with self._mgmt_link_lock:
-            try:
-                mgmt_link = self._mgmt_links[node]
-            except KeyError:
-                mgmt_link = ManagementOperation(self._session, endpoint=node, **kwargs)
-                self._mgmt_links[node] = mgmt_link
-                mgmt_link.open()
+        mgmt_link = self.open_mgmt_link(node=node, timeout=timeout, **kwargs)
 
-        while not self.client_ready():
-            time.sleep(0.05)
-
-        while not mgmt_link.ready():
-            self._connection.listen(wait=False)
         operation_type = operation_type or b"empty"
         status, description, response = mgmt_link.execute(
             message, operation=operation, operation_type=operation_type, timeout=timeout
         )
         return status, description, response
 
+    def open_mgmt_link(self, node: str = "$management", timeout: float = 0, **kwargs):
+        """Open and wait for a management link without dispatching a request.
+
+        :param str node: Management target.
+        :param float timeout: Timeout in seconds.
+        :returns: The opened management link.
+        :rtype: ~pyamqp.management_link.ManagementOperation
+        """
+        start_time = time.monotonic()
+        mgmt_link = None
+        try:
+            with self._mgmt_link_lock:
+                try:
+                    mgmt_link = self._mgmt_links[node]
+                except KeyError:
+                    mgmt_link = ManagementOperation(self._session, endpoint=node, **kwargs)
+                    self._mgmt_links[node] = mgmt_link
+                    mgmt_link.open()
+            while not self.client_ready():
+                if timeout and time.monotonic() - start_time >= timeout:
+                    raise TimeoutError("Management link setup timed out.")
+                time.sleep(0.05)
+
+            while not mgmt_link.ready():
+                if timeout and time.monotonic() - start_time >= timeout:
+                    raise TimeoutError("Management link setup timed out.")
+                self._connection.listen(wait=False)
+            return mgmt_link
+        except Exception:
+            if mgmt_link is not None:
+                with self._mgmt_link_lock:
+                    if self._mgmt_links.get(node) is mgmt_link:
+                        self._mgmt_links.pop(node, None)
+                mgmt_link.close()
+            raise
+
 
 class SendClient(AMQPClient):
     """