Add automatic retry on 429/503 with exponential backoff via urllib3 Retry

Author: bigcat88

README.md

@@ -69,6 +69,7 @@ export NEXTCLOUD_PASSWORD=your-app-password  # Use an app password, not your mai
 
 # Optional
 export NEXTCLOUD_MCP_PERMISSIONS=read  # read (default), write, or destructive
+export NEXTCLOUD_MCP_RETRY_MAX=3       # max retries on 429/503 (default: 3, 0 to disable)
 ```
 
 ### Getting an App Password

pyproject.toml

@@ -106,6 +106,9 @@ venvPath = "."
 venv = "venv"
 reportUnusedFunction = false
 reportPrivateUsage = false
+executionEnvironments = [
+  { root = "tests/test_client_retry.py", reportAttributeAccessIssue = false, reportUnknownMemberType = false }
+]
 
 [tool.pytest]
 ini_options.testpaths = [ "tests" ]

src/nextcloud_mcp/client.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 import niquests
+from urllib3.util import Retry
 
 from .config import Config
 
@@ -70,14 +71,23 @@ def __init__(self, config: Config) -> None:
 
     async def _get_session(self) -> niquests.AsyncSession:
         if self._session is None:
-            self._session = niquests.AsyncSession(
-                auth=(self._config.user, self._config.password),
-                timeout=30,
-                headers={
+            kwargs: dict[str, object] = {
+                "auth": (self._config.user, self._config.password),
+                "timeout": 30,
+                "headers": {
                     "OCS-APIRequest": "true",
                     "Accept": "application/json",
                 },
-            )
+            }
+            if self._config.retry_max > 0:
+                kwargs["retries"] = Retry(
+                    total=self._config.retry_max,
+                    status_forcelist=[429, 503],
+                    backoff_factor=1.0,
+                    respect_retry_after_header=True,
+                    allowed_methods=None,
+                )
+            self._session = niquests.AsyncSession(**kwargs)  # type: ignore[arg-type]
         return self._session
 
     async def close(self) -> None:

src/nextcloud_mcp/config.py

@@ -19,6 +19,7 @@ class Config:
         NEXTCLOUD_MCP_PERMISSIONS: Permission level — 'read' (default), 'write', or 'destructive'
         NEXTCLOUD_MCP_HOST: Host to bind HTTP server (default: 0.0.0.0)
         NEXTCLOUD_MCP_PORT: Port for HTTP server (default: 8100)
+        NEXTCLOUD_MCP_RETRY_MAX: Max retries on 429/503 (default: 3, 0 to disable)
     """
 
     nextcloud_url: str = field(default="")
@@ -27,6 +28,7 @@ class Config:
     permission_level: PermissionLevel = field(default=PermissionLevel.READ)
     host: str = field(default="0.0.0.0")
     port: int = field(default=8100)
+    retry_max: int = field(default=3)
 
     @classmethod
     def from_env(cls) -> "Config":
@@ -44,6 +46,7 @@ def from_env(cls) -> "Config":
 
         host = os.environ.get("NEXTCLOUD_MCP_HOST", "0.0.0.0")
         port = int(os.environ.get("NEXTCLOUD_MCP_PORT", "8100"))
+        retry_max = int(os.environ.get("NEXTCLOUD_MCP_RETRY_MAX", "3"))
 
         return cls(
             nextcloud_url=url,
@@ -52,6 +55,7 @@ def from_env(cls) -> "Config":
             permission_level=perm,
             host=host,
             port=port,
+            retry_max=max(0, retry_max),
         )
 
     def validate(self) -> None:

tests/test_client_retry.py

@@ -0,0 +1,88 @@
+"""Tests for HTTP client retry configuration."""
+
+import pytest
+
+from nextcloud_mcp.client import NextcloudClient
+from nextcloud_mcp.config import Config
+
+
+def _make_config(retry_max: int = 3) -> Config:
+    return Config(
+        nextcloud_url="http://localhost",
+        user="admin",
+        password="admin",
+        retry_max=retry_max,
+    )
+
+
+class TestRetryConfiguration:
+    @pytest.mark.asyncio
+    async def test_retry_enabled_by_default(self) -> None:
+        client = NextcloudClient(_make_config())
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.total == 3
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_status_forcelist(self) -> None:
+        client = NextcloudClient(_make_config())
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert 429 in adapter.max_retries.status_forcelist
+            assert 503 in adapter.max_retries.status_forcelist
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_respects_retry_after(self) -> None:
+        client = NextcloudClient(_make_config())
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.respect_retry_after_header is True
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_custom_max(self) -> None:
+        client = NextcloudClient(_make_config(retry_max=5))
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.total == 5
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_disabled_when_zero(self) -> None:
+        client = NextcloudClient(_make_config(retry_max=0))
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.total == 0
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_allows_all_methods(self) -> None:
+        client = NextcloudClient(_make_config())
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.allowed_methods is None
+        finally:
+            await client.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_backoff_factor(self) -> None:
+        client = NextcloudClient(_make_config())
+        session = await client._get_session()
+        try:
+            adapter = session.get_adapter("http://localhost")
+            assert adapter.max_retries.backoff_factor == 1.0
+        finally:
+            await client.close()

tests/test_config.py

@@ -60,6 +60,41 @@ def test_invalid_permission_raises(self, monkeypatch: pytest.MonkeyPatch) -> Non
         with pytest.raises(ValueError, match="Invalid NEXTCLOUD_MCP_PERMISSIONS"):
             Config.from_env()
 
+    def test_default_retry_max(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("NEXTCLOUD_URL", "http://localhost")
+        monkeypatch.setenv("NEXTCLOUD_USER", "admin")
+        monkeypatch.setenv("NEXTCLOUD_PASSWORD", "admin")
+
+        config = Config.from_env()
+        assert config.retry_max == 3
+
+    def test_retry_max_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("NEXTCLOUD_URL", "http://localhost")
+        monkeypatch.setenv("NEXTCLOUD_USER", "admin")
+        monkeypatch.setenv("NEXTCLOUD_PASSWORD", "admin")
+        monkeypatch.setenv("NEXTCLOUD_MCP_RETRY_MAX", "5")
+
+        config = Config.from_env()
+        assert config.retry_max == 5
+
+    def test_retry_max_zero_disables(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("NEXTCLOUD_URL", "http://localhost")
+        monkeypatch.setenv("NEXTCLOUD_USER", "admin")
+        monkeypatch.setenv("NEXTCLOUD_PASSWORD", "admin")
+        monkeypatch.setenv("NEXTCLOUD_MCP_RETRY_MAX", "0")
+
+        config = Config.from_env()
+        assert config.retry_max == 0
+
+    def test_retry_max_negative_clamped_to_zero(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("NEXTCLOUD_URL", "http://localhost")
+        monkeypatch.setenv("NEXTCLOUD_USER", "admin")
+        monkeypatch.setenv("NEXTCLOUD_PASSWORD", "admin")
+        monkeypatch.setenv("NEXTCLOUD_MCP_RETRY_MAX", "-1")
+
+        config = Config.from_env()
+        assert config.retry_max == 0
+
     def test_case_insensitive_permission(self, monkeypatch: pytest.MonkeyPatch) -> None:
         monkeypatch.setenv("NEXTCLOUD_URL", "http://localhost")
         monkeypatch.setenv("NEXTCLOUD_USER", "admin")