fix(python): caching on resource loader

python/pyproject.toml

@@ -18,10 +18,11 @@ dependencies = [
     "opentelemetry-instrumentation-fastapi>=0.52b1",
     "opentelemetry-sdk>=1.31",
     "janus>=2.0",
-    "cachetools>=5.5",
     "redis>=6.1",
     "psycopg[binary]>=3.2",
     "obstore>=0.6",
+    "async-lru>=2.0.5",
+    "cachetools>=6.1.0",
 ]
 
 [build-system]

python/src/acp_sdk/server/resources.py

@@ -5,7 +5,7 @@
 import uuid
 from typing import Callable
 
-import cachetools
+from async_lru import alru_cache
 
 from acp_sdk.models import ResourceUrl
 from acp_sdk.models.types import ResourceId
@@ -30,7 +30,7 @@ def __init__(
             else None
         )
 
-    @cachetools.func.lfu_cache
+    @alru_cache()
     async def load(self, url: ResourceUrl) -> bytes:
         if self._url_pattern:
             match = re.match(self._url_pattern, str(url))

python/src/acp_sdk/shared/resources.py

@@ -3,10 +3,9 @@
 
 from datetime import timedelta
 
-import cachetools
-import cachetools.func
 import httpx
 import obstore
+from async_lru import alru_cache
 from obstore.store import AzureStore, GCSStore, HTTPStore, ObjectStore, S3Store
 
 from acp_sdk.models.types import ResourceId, ResourceUrl
@@ -16,7 +15,7 @@ class ResourceLoader:
     def __init__(self, *, client: httpx.AsyncClient | None = None) -> None:
         self._client = client or httpx.AsyncClient(follow_redirects=False)
 
-    @cachetools.func.lfu_cache
+    @alru_cache()
     async def load(self, url: ResourceUrl) -> bytes:
         response = await self._client.get(str(url))
         response.raise_for_status()

python/tests/unit/shared/test_resources.py

@@ -0,0 +1,29 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+
+import pytest
+from acp_sdk.shared.resources import ResourceLoader
+from pytest_httpx import HTTPXMock
+
+mock_resource = {"url": "http://invalid/resource1", "content": b"foobar"}
+
+
+@pytest.mark.asyncio
+async def test_resource_loader_load(httpx_mock: HTTPXMock) -> None:
+    httpx_mock.add_response(url=mock_resource["url"], method="GET", content=mock_resource["content"])
+
+    resource_loader = ResourceLoader()
+    resource = await resource_loader.load(mock_resource["url"])
+    assert resource == mock_resource["content"]
+
+
+@pytest.mark.asyncio
+async def test_resource_loader_load_cache(httpx_mock: HTTPXMock) -> None:
+    httpx_mock.add_response(url=mock_resource["url"], method="GET", content=mock_resource["content"])
+
+    resource_loader = ResourceLoader()
+    resource = await resource_loader.load(mock_resource["url"])
+    resource = await resource_loader.load(mock_resource["url"])
+
+    assert resource == mock_resource["content"]