@@ -11,6 +11,17 @@ class RetryHandlerOption(RequestOption):
         max_retries (int): number of retries after the first attempt, 0 to
             ``MAX_MAX_RETRIES``. Exposed as ``max_retry``.
         should_retry (bool): whether the handler retries at all.
+        retry_unprocessed_mutations_only (bool): whether POST and PATCH requests are retried
+            only after a response showing the server didn't process them. Defaults to False.
+            POST and PATCH aren't idempotent, so resending one the server already processed
+            can repeat its side effect, such as sending an email twice. A 504 can arrive after
+            the server processed the request. When this is True, POST and PATCH requests are
+            retried only after a 429 or 503 response, and a 504 is returned to the caller.
+            This relies on the common service convention, followed by Microsoft Graph, that a
+            429 or 503 response means the server didn't process the request. HTTP doesn't
+            guarantee this, so a service that returns 429 or 503 after acting on a request can
+            still see it repeated. PUT and DELETE also change state, but they're idempotent, so
+            they aren't affected, and neither are other methods.
     """
 
     # Default maxRetries value
@@ -35,7 +46,8 @@ def __init__(
         self,
         delay: float = DEFAULT_DELAY,
         max_retries: int = DEFAULT_MAX_RETRIES,
-        should_retry: bool = DEFAULT_SHOULD_RETRY
+        should_retry: bool = DEFAULT_SHOULD_RETRY,
+        retry_unprocessed_mutations_only: bool = False
     ) -> None:
         if delay > self.MAX_DELAY and max_retries > self.DEFAULT_MAX_RETRIES:
             raise ValueError(
@@ -58,6 +70,7 @@ def __init__(
         self._max_retry: int = min(max_retries, self.MAX_MAX_RETRIES)
         self._max_delay: float = min(delay, self.MAX_DELAY)
         self._should_retry: bool = should_retry
+        self._retry_unprocessed_mutations_only: bool = retry_unprocessed_mutations_only
 
     @property
     def max_delay(self) -> float:
@@ -93,6 +106,14 @@ def should_retry(self) -> bool:
     def should_retry(self, value: bool) -> None:
         self._should_retry = value
 
+    @property
+    def retry_unprocessed_mutations_only(self) -> bool:
+        return self._retry_unprocessed_mutations_only
+
+    @retry_unprocessed_mutations_only.setter
+    def retry_unprocessed_mutations_only(self, value: bool) -> None:
+        self._retry_unprocessed_mutations_only = value
+
     @staticmethod
     def get_key() -> str:
         return RetryHandlerOption.RETRY_HANDLER_OPTION_KEY
@@ -43,6 +43,15 @@ class RetryHandler(BaseMiddleware):
         ['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS', 'QUERY']
     )
 
+    # Methods that aren't idempotent, so a retry can repeat their side effect
+    NON_IDEMPOTENT_METHODS: frozenset[str] = frozenset(['POST', 'PATCH'])
+
+    # By common service convention, these status codes mean the server didn't process
+    # the request, so resending it is treated as safe for any method
+    # 429 - Too many requests
+    # 503 - Service unavailable
+    UNPROCESSED_STATUS_CODES: frozenset[int] = frozenset({429, 503})
+
     def __init__(self, options: Optional[RetryHandlerOption] = None) -> None:
         super().__init__()
         self.allowed_methods: frozenset[str] = self.DEFAULT_ALLOWED_METHODS
@@ -129,8 +138,25 @@ def should_retry(
             return False
         if not self._is_request_payload_buffered(request):
             return False
+        if not self._is_retryable_for_method(request, options, response):
+            return False
         return options.max_retry > 0 and response.status_code in self.retry_on_status_codes
 
+    def _is_retryable_for_method(
+        self, request: httpx.Request, options: RetryHandlerOption, response: httpx.Response
+    ) -> bool:
+        """
+        Checks whether the response allows resending the request, given its method.
+        With retry_unprocessed_mutations_only set, POST and PATCH requests are retried
+        only after a response showing the server didn't process them, because any other
+        response may arrive after the server already acted on the request.
+        """
+        if not options.retry_unprocessed_mutations_only:
+            return True
+        if request.method.upper() not in self.NON_IDEMPOTENT_METHODS:
+            return True
+        return response.status_code in self.UNPROCESSED_STATUS_CODES
+
     def _is_method_retryable(self, request: httpx.Request) -> bool:
         """
         Checks if a given request should be retried upon, depending on
 
@@ -9,6 +9,7 @@
     PARENT_SPAN_KEY,
     REQUEST_OPTIONS_KEY,
     MiddlewarePipeline,
+    RedirectHandler,
     RetryHandler,
     UserAgentHandler,
 )
@@ -444,3 +445,175 @@ def request_handler(request: httpx.Request):
     assert len(ua_spans) == 2
     for ua_span in ua_spans:
         assert ua_span.parent.span_id == expected_parent_id
+
+
+def _status_sequence_transport(status_codes, methods):
+    """Answers each request with the next status code and records its method."""
+    remaining = list(status_codes)
+
+    def request_handler(request: httpx.Request):
+        methods.append(request.method)
+        return httpx.Response(remaining.pop(0) if remaining else 200)
+
+    return httpx.MockTransport(request_handler)
+
+
+@pytest.fixture
+def no_retry_delay(monkeypatch):
+    """Skips the wait between retries so the tests run quickly."""
+    monkeypatch.setattr(RetryHandler, "get_delay_time", lambda self, *args, **kwargs: 0)
+
+
+def test_retry_unprocessed_mutations_only_defaults_to_false():
+    assert not RetryHandlerOption().retry_unprocessed_mutations_only
+    assert RetryHandlerOption(
+        retry_unprocessed_mutations_only=True
+    ).retry_unprocessed_mutations_only
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["POST", "PATCH"])
+async def test_non_idempotent_request_retried_on_gateway_timeout_by_default(method, no_retry_delay):
+    methods = []
+    handler = RetryHandler()
+    request = httpx.Request(method, BASE_URL, json={"message": "hello"})
+
+    resp = await handler.send(request, _status_sequence_transport([GATEWAY_TIMEOUT, 200], methods))
+
+    assert resp.status_code == 200
+    assert len(methods) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["POST", "PATCH", "post"])
+async def test_non_idempotent_request_not_retried_on_gateway_timeout_when_enabled(
+    method, no_retry_delay
+):
+    methods = []
+    handler = RetryHandler(RetryHandlerOption(retry_unprocessed_mutations_only=True))
+    request = httpx.Request(method, BASE_URL, json={"message": "hello"})
+
+    resp = await handler.send(request, _status_sequence_transport([GATEWAY_TIMEOUT, 200], methods))
+
+    assert resp.status_code == GATEWAY_TIMEOUT
+    assert len(methods) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["POST", "PATCH"])
+@pytest.mark.parametrize("status_code", [TOO_MANY_REQUESTS, SERVICE_UNAVAILABLE])
+async def test_non_idempotent_request_retried_when_not_processed_and_enabled(
+    method, status_code, no_retry_delay
+):
+    methods = []
+    handler = RetryHandler(RetryHandlerOption(retry_unprocessed_mutations_only=True))
+    request = httpx.Request(method, BASE_URL, json={"message": "hello"})
+
+    resp = await handler.send(request, _status_sequence_transport([status_code, 200], methods))
+
+    assert resp.status_code == 200
+    assert len(methods) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["GET", "PUT", "DELETE", "QUERY"])
+async def test_idempotent_request_retried_on_gateway_timeout_when_enabled(method, no_retry_delay):
+    methods = []
+    handler = RetryHandler(RetryHandlerOption(retry_unprocessed_mutations_only=True))
+    request = httpx.Request(method, BASE_URL)
+
+    resp = await handler.send(request, _status_sequence_transport([GATEWAY_TIMEOUT, 200], methods))
+
+    assert resp.status_code == 200
+    assert len(methods) == 2
+
+
+@pytest.mark.asyncio
+async def test_non_idempotent_request_stops_when_a_later_response_leaves_outcome_unknown(
+    no_retry_delay
+):
+    methods = []
+    handler = RetryHandler(RetryHandlerOption(retry_unprocessed_mutations_only=True))
+    request = httpx.Request("POST", BASE_URL, json={"message": "hello"})
+
+    resp = await handler.send(
+        request, _status_sequence_transport([SERVICE_UNAVAILABLE, GATEWAY_TIMEOUT, 200], methods)
+    )
+
+    assert resp.status_code == GATEWAY_TIMEOUT
+    assert len(methods) == 2
+
+
+@pytest.mark.asyncio
+async def test_retry_unprocessed_mutations_only_honoured_from_request_option(no_retry_delay):
+    methods = []
+    handler = RetryHandler()
+    request_option = RetryHandlerOption(retry_unprocessed_mutations_only=True)
+    request = httpx.Request(
+        "POST",
+        BASE_URL,
+        json={"message": "hello"},
+        extensions={REQUEST_OPTIONS_KEY: {
+            RetryHandlerOption.get_key(): request_option
+        }},
+    )
+
+    resp = await handler.send(request, _status_sequence_transport([GATEWAY_TIMEOUT, 200], methods))
+
+    assert resp.status_code == GATEWAY_TIMEOUT
+    assert len(methods) == 1
+
+
+def _redirect_then_status_transport(redirect_status, final_statuses, calls):
+    """Redirects the first request to /redirected, then answers with each final status."""
+    remaining = list(final_statuses)
+
+    def request_handler(request: httpx.Request):
+        calls.append(request.method)
+        if request.url.path != "/redirected":
+            return httpx.Response(redirect_status, headers={"Location": f"{BASE_URL}/redirected"})
+        return httpx.Response(remaining.pop(0) if remaining else 200)
+
+    return httpx.MockTransport(request_handler)
+
+
+@pytest.mark.asyncio
+async def test_post_redirected_to_get_is_retried_when_enabled(no_retry_delay):
+    calls = []
+    pipeline = MiddlewarePipeline(
+        _redirect_then_status_transport(303, [GATEWAY_TIMEOUT, 200], calls)
+    )
+    pipeline.add_middleware(RedirectHandler())
+    pipeline.add_middleware(RetryHandler(RetryHandlerOption(retry_unprocessed_mutations_only=True)))
+    request = httpx.Request("POST", f"{BASE_URL}/send", json={"message": "hello"})
+
+    resp = await pipeline.send(request)
+
+    assert resp.status_code == 200
+    assert calls == ["POST", "GET", "GET"]
+
+
+@pytest.mark.asyncio
+async def test_post_redirected_as_post_is_not_retried_with_request_option(no_retry_delay):
+    calls = []
+    pipeline = MiddlewarePipeline(
+        _redirect_then_status_transport(307, [GATEWAY_TIMEOUT, 200], calls)
+    )
+    pipeline.add_middleware(RedirectHandler())
+    pipeline.add_middleware(RetryHandler())
+    request = httpx.Request(
+        "POST",
+        f"{BASE_URL}/send",
+        json={"message": "hello"},
+        extensions={
+            REQUEST_OPTIONS_KEY: {
+                RetryHandlerOption.get_key():
+                RetryHandlerOption(retry_unprocessed_mutations_only=True)
+            }
+        },
+    )
+
+    resp = await pipeline.send(request)
+
+    assert resp.status_code == GATEWAY_TIMEOUT
+    assert calls == ["POST", "POST"]