diff --git a/lib/crewai-files/src/crewai_files/cache/upload_cache.py b/lib/crewai-files/src/crewai_files/cache/upload_cache.py
index 48cebdfa14..c94e164c7e 100644
--- a/lib/crewai-files/src/crewai_files/cache/upload_cache.py
+++ b/lib/crewai-files/src/crewai_files/cache/upload_cache.py
@@ -1,4 +1,4 @@
-"""Cache for tracking uploaded files using aiocache."""
+"""Cache for tracking uploaded files using aiocache or ValkeyCache."""
 
 from __future__ import annotations
 
@@ -10,10 +10,11 @@
 from datetime import datetime, timezone
 import hashlib
 import logging
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol
 
 from aiocache import Cache  # type: ignore[import-untyped]
 from aiocache.serializers import PickleSerializer  # type: ignore[import-untyped]
+from crewai.utilities.cache_config import parse_cache_url
 
 from crewai_files.core.constants import DEFAULT_MAX_CACHE_ENTRIES, DEFAULT_TTL_SECONDS
 from crewai_files.uploaders.factory import ProviderType
@@ -51,6 +52,33 @@ def is_expired(self) -> bool:
             return False
         return datetime.now(timezone.utc) >= self.expires_at
 
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize to a JSON-compatible dict."""
+        return {
+            "file_id": self.file_id,
+            "provider": self.provider,
+            "file_uri": self.file_uri,
+            "content_type": self.content_type,
+            "uploaded_at": self.uploaded_at.isoformat(),
+            "expires_at": self.expires_at.isoformat() if self.expires_at else None,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> CachedUpload:
+        """Deserialize from a dict."""
+        return cls(
+            file_id=data["file_id"],
+            provider=data["provider"],
+            file_uri=data.get("file_uri"),
+            content_type=data["content_type"],
+            uploaded_at=datetime.fromisoformat(data["uploaded_at"]),
+            expires_at=(
+                datetime.fromisoformat(data["expires_at"])
+                if data.get("expires_at")
+                else None
+            ),
+        )
+
 
 def _make_key(file_hash: str, provider: str) -> str:
     """Create a cache key from file hash and provider."""
@@ -58,14 +86,7 @@ def _make_key(file_hash: str, provider: str) -> str:
 
 
 def _compute_file_hash_streaming(chunks: Iterator[bytes]) -> str:
-    """Compute SHA-256 hash from streaming chunks.
-
-    Args:
-        chunks: Iterator of byte chunks.
-
-    Returns:
-        Hexadecimal hash string.
-    """
+    """Compute SHA-256 hash from streaming chunks."""
     hasher = hashlib.sha256()
     for chunk in chunks:
         hasher.update(chunk)
@@ -73,10 +94,7 @@ def _compute_file_hash_streaming(chunks: Iterator[bytes]) -> str:
 
 
 def _compute_file_hash(file: FileInput) -> str:
-    """Compute SHA-256 hash of file content.
-
-    Uses streaming for FilePath sources to avoid loading large files into memory.
-    """
+    """Compute SHA-256 hash of file content."""
     from crewai_files.core.sources import FilePath
 
     source = file._file_source
@@ -86,10 +104,73 @@ def _compute_file_hash(file: FileInput) -> str:
     return hashlib.sha256(content).hexdigest()
 
 
+class CacheBackend(Protocol):
+    """Protocol for cache backends used by UploadCache."""
+
+    async def get(self, key: str) -> CachedUpload | None: ...
+    async def set(self, key: str, value: CachedUpload, ttl: int) -> None: ...
+    async def delete(self, key: str) -> bool: ...
+
+
+class AiocacheBackend:
+    """Cache backend backed by aiocache (memory or Redis)."""
+
+    def __init__(self, cache: Cache) -> None:  # type: ignore[no-any-unimported]
+        self._cache = cache
+
+    async def get(self, key: str) -> CachedUpload | None:
+        result = await self._cache.get(key)
+        if isinstance(result, CachedUpload):
+            return result
+        return None
+
+    async def set(self, key: str, value: CachedUpload, ttl: int) -> None:
+        await self._cache.set(key, value, ttl=ttl)
+
+    async def delete(self, key: str) -> bool:
+        result = await self._cache.delete(key)
+        return bool(result > 0 if isinstance(result, int) else result)
+
+
+class ValkeyCacheBackend:
+    """Cache backend backed by ValkeyCache (JSON serialization)."""
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        db: int = 0,
+        password: str | None = None,
+        default_ttl: int | None = None,
+    ) -> None:
+        from crewai.memory.storage.valkey_cache import ValkeyCache
+
+        self._cache = ValkeyCache(
+            host=host, port=port, db=db, password=password, default_ttl=default_ttl
+        )
+
+    async def get(self, key: str) -> CachedUpload | None:
+        data = await self._cache.get(key)
+        if data is None:
+            return None
+        try:
+            return CachedUpload.from_dict(data)
+        except (KeyError, ValueError) as e:
+            logger.warning(f"Failed to deserialize cached upload: {e}")
+            return None
+
+    async def set(self, key: str, value: CachedUpload, ttl: int) -> None:
+        await self._cache.set(key, value.to_dict(), ttl=ttl)
+
+    async def delete(self, key: str) -> bool:
+        await self._cache.delete(key)
+        return True  # ValkeyCache.delete is void
+
+
 class UploadCache:
-    """Async cache for tracking uploaded files using aiocache.
+    """Async cache for tracking uploaded files.
 
-    Supports in-memory caching by default, with optional Redis backend
+    Supports in-memory caching by default, with optional Redis or Valkey backend
     for distributed setups.
 
     Attributes:
@@ -110,7 +191,7 @@ def __init__(
         Args:
             ttl: Default TTL in seconds.
             namespace: Cache namespace.
-            cache_type: Backend type ("memory" or "redis").
+            cache_type: Backend type ("memory", "redis", or "valkey").
             max_entries: Maximum cache entries (None for unlimited).
             **cache_kwargs: Additional args for cache backend.
         """
@@ -120,18 +201,39 @@ def __init__(
         self._provider_keys: dict[ProviderType, set[str]] = {}
         self._key_access_order: list[str] = []
 
-        if cache_type == "redis":
-            self._cache = Cache(
-                Cache.REDIS,
-                serializer=PickleSerializer(),
-                namespace=namespace,
-                **cache_kwargs,
+        self._backend: CacheBackend = self._create_backend(
+            cache_type, namespace, ttl, **cache_kwargs
+        )
+
+    @staticmethod
+    def _create_backend(
+        cache_type: str,
+        namespace: str,
+        ttl: int,
+        **cache_kwargs: Any,
+    ) -> CacheBackend:
+        """Create the appropriate cache backend."""
+        if cache_type == "valkey":
+            conn = parse_cache_url() or {}
+            return ValkeyCacheBackend(
+                host=cache_kwargs.get("host", conn.get("host", "localhost")),
+                port=cache_kwargs.get("port", conn.get("port", 6379)),
+                db=cache_kwargs.get("db", conn.get("db", 0)),
+                password=cache_kwargs.get("password", conn.get("password")),
+                default_ttl=ttl,
             )
-        else:
-            self._cache = Cache(
-                serializer=PickleSerializer(),
-                namespace=namespace,
+        if cache_type == "redis":
+            return AiocacheBackend(
+                Cache(
+                    Cache.REDIS,
+                    serializer=PickleSerializer(),
+                    namespace=namespace,
+                    **cache_kwargs,
+                )
             )
+        return AiocacheBackend(
+            Cache(serializer=PickleSerializer(), namespace=namespace)
+        )
 
     def _track_key(self, provider: ProviderType, key: str) -> None:
         """Track a key for a provider (for cleanup) and access order."""
@@ -157,11 +259,9 @@ async def _evict_if_needed(self) -> int:
         """
         if self.max_entries is None:
             return 0
-
         current_count = len(self)
         if current_count < self.max_entries:
             return 0
-
         to_evict = max(1, self.max_entries // 10)
         return await self._evict_oldest(to_evict)
 
@@ -176,31 +276,24 @@ async def _evict_oldest(self, count: int) -> int:
         """
         evicted = 0
         keys_to_evict = self._key_access_order[:count]
-
         for key in keys_to_evict:
-            await self._cache.delete(key)
+            await self._backend.delete(key)
             self._key_access_order.remove(key)
             for provider_keys in self._provider_keys.values():
                 provider_keys.discard(key)
             evicted += 1
-
         if evicted > 0:
             logger.debug(f"Evicted {evicted} oldest cache entries")
-
         return evicted
 
+    # ------------------------------------------------------------------
+    # Async public API
+    # ------------------------------------------------------------------
+
     async def aget(
         self, file: FileInput, provider: ProviderType
     ) -> CachedUpload | None:
-        """Get a cached upload for a file.
-
-        Args:
-            file: The file to look up.
-            provider: The provider name.
-
-        Returns:
-            Cached upload if found and not expired, None otherwise.
-        """
+        """Get a cached upload for a file."""
         file_hash = _compute_file_hash(file)
         return await self.aget_by_hash(file_hash, provider)
 
@@ -217,17 +310,14 @@ async def aget_by_hash(
             Cached upload if found and not expired, None otherwise.
         """
         key = _make_key(file_hash, provider)
-        result = await self._cache.get(key)
-
+        result = await self._backend.get(key)
         if result is None:
             return None
-        if isinstance(result, CachedUpload):
-            if result.is_expired():
-                await self._cache.delete(key)
-                self._untrack_key(provider, key)
-                return None
-            return result
-        return None
+        if result.is_expired():
+            await self._backend.delete(key)
+            self._untrack_key(provider, key)
+            return None
+        return result
 
     async def aset(
         self,
@@ -237,18 +327,7 @@ async def aset(
         file_uri: str | None = None,
         expires_at: datetime | None = None,
     ) -> CachedUpload:
-        """Cache an uploaded file.
-
-        Args:
-            file: The file that was uploaded.
-            provider: The provider name.
-            file_id: Provider-specific file identifier.
-            file_uri: Optional URI for accessing the file.
-            expires_at: When the upload expires.
-
-        Returns:
-            The created cache entry.
-        """
+        """Cache an uploaded file."""
         file_hash = _compute_file_hash(file)
         return await self.aset_by_hash(
             file_hash=file_hash,
@@ -282,7 +361,6 @@ async def aset_by_hash(
             The created cache entry.
         """
         await self._evict_if_needed()
-
         key = _make_key(file_hash, provider)
         now = datetime.now(timezone.utc)
 
@@ -299,7 +377,7 @@ async def aset_by_hash(
         if expires_at is not None:
             ttl = max(0, int((expires_at - now).total_seconds()))
 
-        await self._cache.set(key, cached, ttl=ttl)
+        await self._backend.set(key, cached, ttl=ttl)
         self._track_key(provider, key)
         logger.debug(f"Cached upload: {file_id} for provider {provider}")
         return cached
@@ -316,9 +394,7 @@ async def aremove(self, file: FileInput, provider: ProviderType) -> bool:
         """
         file_hash = _compute_file_hash(file)
         key = _make_key(file_hash, provider)
-
-        result = await self._cache.delete(key)
-        removed = bool(result > 0 if isinstance(result, int) else result)
+        removed = await self._backend.delete(key)
         if removed:
             self._untrack_key(provider, key)
         return removed
@@ -335,11 +411,10 @@ async def aremove_by_file_id(self, file_id: str, provider: ProviderType) -> bool
         """
         if provider not in self._provider_keys:
             return False
-
         for key in list(self._provider_keys[provider]):
-            cached = await self._cache.get(key)
-            if isinstance(cached, CachedUpload) and cached.file_id == file_id:
-                await self._cache.delete(key)
+            cached = await self._backend.get(key)
+            if cached is not None and cached.file_id == file_id:
+                await self._backend.delete(key)
                 self._untrack_key(provider, key)
                 return True
         return False
@@ -351,17 +426,13 @@ async def aclear_expired(self) -> int:
             Number of entries removed.
         """
         removed = 0
-
         for provider, keys in list(self._provider_keys.items()):
             for key in list(keys):
-                cached = await self._cache.get(key)
-                if cached is None or (
-                    isinstance(cached, CachedUpload) and cached.is_expired()
-                ):
-                    await self._cache.delete(key)
+                cached = await self._backend.get(key)
+                if cached is None or cached.is_expired():
+                    await self._backend.delete(key)
                     self._untrack_key(provider, key)
                     removed += 1
-
         if removed > 0:
             logger.debug(f"Cleared {removed} expired cache entries")
         return removed
@@ -373,9 +444,12 @@ async def aclear(self) -> int:
             Number of entries cleared.
         """
         count = sum(len(keys) for keys in self._provider_keys.values())
-        await self._cache.clear(namespace=self.namespace)
+        # Delete all tracked keys individually (works for all backends)
+        for keys in self._provider_keys.values():
+            for key in keys:
+                await self._backend.delete(key)
         self._provider_keys.clear()
-
+        self._key_access_order.clear()
         if count > 0:
             logger.debug(f"Cleared {count} cache entries")
         return count
@@ -391,14 +465,17 @@ async def aget_all_for_provider(self, provider: ProviderType) -> list[CachedUplo
         """
         if provider not in self._provider_keys:
             return []
-
         results: list[CachedUpload] = []
         for key in list(self._provider_keys[provider]):
-            cached = await self._cache.get(key)
-            if isinstance(cached, CachedUpload) and not cached.is_expired():
+            cached = await self._backend.get(key)
+            if cached is not None and not cached.is_expired():
                 results.append(cached)
         return results
 
+    # ------------------------------------------------------------------
+    # Sync wrappers
+    # ------------------------------------------------------------------
+
     @staticmethod
     def _run_sync(coro: Any) -> Any:
         """Run an async coroutine from sync context without blocking event loop."""
@@ -489,11 +566,7 @@ def __len__(self) -> int:
         return sum(len(keys) for keys in self._provider_keys.values())
 
     def get_providers(self) -> builtins.set[ProviderType]:
-        """Get all provider names that have cached entries.
-
-        Returns:
-            Set of provider names.
-        """
+        """Get all provider names that have cached entries."""
         return builtins.set(self._provider_keys.keys())
 
 
@@ -506,17 +579,7 @@ def get_upload_cache(
     cache_type: str = "memory",
     **cache_kwargs: Any,
 ) -> UploadCache:
-    """Get or create the default upload cache.
-
-    Args:
-        ttl: Default TTL in seconds.
-        namespace: Cache namespace.
-        cache_type: Backend type ("memory" or "redis").
-        **cache_kwargs: Additional args for cache backend.
-
-    Returns:
-        The upload cache instance.
-    """
+    """Get or create the default upload cache."""
     global _default_cache
     if _default_cache is None:
         _default_cache = UploadCache(
diff --git a/lib/crewai/pyproject.toml b/lib/crewai/pyproject.toml
index 741c53b558..1abed0651d 100644
--- a/lib/crewai/pyproject.toml
+++ b/lib/crewai/pyproject.toml
@@ -110,6 +110,9 @@ file-processing = [
 qdrant-edge = [
     "qdrant-edge-py>=0.6.0",
 ]
+valkey = [
+    "valkey-glide>=1.3.0",
+]
 
 
 [tool.uv]
diff --git a/lib/crewai/src/crewai/a2a/utils/agent_card.py b/lib/crewai/src/crewai/a2a/utils/agent_card.py
index df5886988e..d3a47e2fef 100644
--- a/lib/crewai/src/crewai/a2a/utils/agent_card.py
+++ b/lib/crewai/src/crewai/a2a/utils/agent_card.py
@@ -13,8 +13,12 @@
 from typing import TYPE_CHECKING
 
 from a2a.client.errors import A2AClientHTTPError
-from a2a.types import AgentCapabilities, AgentCard, AgentSkill
-from aiocache import cached  # type: ignore[import-untyped]
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentSkill,
+)
+from aiocache import cached, caches  # type: ignore[import-untyped]
 from aiocache.serializers import PickleSerializer  # type: ignore[import-untyped]
 import httpx
 
@@ -32,6 +36,7 @@
     A2AAuthenticationFailedEvent,
     A2AConnectionErrorEvent,
 )
+from crewai.utilities.cache_config import get_aiocache_config
 
 
 if TYPE_CHECKING:
@@ -40,6 +45,18 @@
     from crewai.task import Task
 
 
+_cache_configured = False
+
+
+def _ensure_cache_configured() -> None:
+    """Configure aiocache on first use (lazy initialization)."""
+    global _cache_configured
+    if _cache_configured:
+        return
+    caches.set_config(get_aiocache_config())
+    _cache_configured = True
+
+
 def _get_tls_verify(auth: ClientAuthScheme | None) -> ssl.SSLContext | bool | str:
     """Get TLS verify parameter from auth scheme.
 
@@ -191,6 +208,7 @@ async def afetch_agent_card(
         else:
             auth_hash = _auth_store.compute_key("none", "")
         _auth_store.set(auth_hash, auth)
+        _ensure_cache_configured()
         agent_card: AgentCard = await _afetch_agent_card_cached(
             endpoint, auth_hash, timeout
         )
diff --git a/lib/crewai/src/crewai/a2a/utils/task.py b/lib/crewai/src/crewai/a2a/utils/task.py
index 6af935bb35..478c5c5f8a 100644
--- a/lib/crewai/src/crewai/a2a/utils/task.py
+++ b/lib/crewai/src/crewai/a2a/utils/task.py
@@ -9,9 +9,8 @@
 from functools import wraps
 import json
 import logging
-import os
+import threading
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
-from urllib.parse import urlparse
 
 from a2a.server.agent_execution import RequestContext
 from a2a.server.events import EventQueue
@@ -38,7 +37,6 @@
 from a2a.utils.errors import ServerError
 from aiocache import SimpleMemoryCache, caches  # type: ignore[import-untyped]
 from pydantic import BaseModel
-from typing_extensions import TypedDict
 
 from crewai.a2a.utils.agent_card import _get_server_config
 from crewai.a2a.utils.content_type import validate_message_parts
@@ -50,12 +48,18 @@
     A2AServerTaskStartedEvent,
 )
 from crewai.task import Task
+from crewai.utilities.cache_config import (
+    get_aiocache_config,
+    parse_cache_url,
+    use_valkey_cache,
+)
 from crewai.utilities.pydantic_schema_utils import create_model_from_schema
 
 
 if TYPE_CHECKING:
     from crewai.a2a.extensions.server import ExtensionContext, ServerExtensionRegistry
     from crewai.agent import Agent
+    from crewai.memory.storage.valkey_cache import ValkeyCache
 
 
 logger = logging.getLogger(__name__)
@@ -64,52 +68,49 @@
 T = TypeVar("T")
 
 
-class RedisCacheConfig(TypedDict, total=False):
-    """Configuration for aiocache Redis backend."""
+# ---------------------------------------------------------------------------
+# Lazy cache initialisation
+# ---------------------------------------------------------------------------
 
-    cache: str
-    endpoint: str
-    port: int
-    db: int
-    password: str
+_task_cache: ValkeyCache | None = None
+_cache_initialized = False
+_cache_init_lock = threading.Lock()
 
+# Configure aiocache at import time (matches upstream behaviour).
+# This is safe — it only touches aiocache, no optional dependencies.
+# The Valkey path is deferred to _ensure_task_cache() to avoid importing
+# valkey-glide at module level (it may not be installed).
+if not use_valkey_cache():
+    caches.set_config(get_aiocache_config())
 
-def _parse_redis_url(url: str) -> RedisCacheConfig:
-    """Parse a Redis URL into aiocache configuration.
 
-    Args:
-        url: Redis connection URL (e.g., redis://localhost:6379/0).
+def _ensure_task_cache() -> None:
+    """Initialise the Valkey task cache on first use (thread-safe).
 
-    Returns:
-        Configuration dict for aiocache.RedisCache.
+    For the aiocache path, configuration happens at module level above.
+    This function only needs to run for the Valkey path.
     """
-    parsed = urlparse(url)
-    config: RedisCacheConfig = {
-        "cache": "aiocache.RedisCache",
-        "endpoint": parsed.hostname or "localhost",
-        "port": parsed.port or 6379,
-    }
-    if parsed.path and parsed.path != "/":
-        try:
-            config["db"] = int(parsed.path.lstrip("/"))
-        except ValueError:
-            pass
-    if parsed.password:
-        config["password"] = parsed.password
-    return config
-
+    global _task_cache, _cache_initialized
+    if _cache_initialized:
+        return
+
+    with _cache_init_lock:
+        if _cache_initialized:
+            return
+
+        if use_valkey_cache():
+            from crewai.memory.storage.valkey_cache import ValkeyCache
+
+            conn = parse_cache_url() or {}
+            _task_cache = ValkeyCache(
+                host=conn.get("host", "localhost"),
+                port=conn.get("port", 6379),
+                db=conn.get("db", 0),
+                password=conn.get("password"),
+                default_ttl=3600,
+            )
 
-_redis_url = os.environ.get("REDIS_URL")
-
-caches.set_config(
-    {
-        "default": _parse_redis_url(_redis_url)
-        if _redis_url
-        else {
-            "cache": "aiocache.SimpleMemoryCache",
-        }
-    }
-)
+        _cache_initialized = True
 
 
 def cancellable(
@@ -130,6 +131,8 @@ def cancellable(
     @wraps(fn)
     async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
         """Wrap function with cancellation monitoring."""
+        _ensure_task_cache()
+
         context: RequestContext | None = None
         for arg in args:
             if isinstance(arg, RequestContext):
@@ -142,10 +145,19 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
             return await fn(*args, **kwargs)
 
         task_id = context.task_id
-        cache = caches.get("default")
 
-        async def poll_for_cancel() -> bool:
-            """Poll cache for cancellation flag."""
+        async def poll_for_cancel_valkey() -> bool:
+            """Poll ValkeyCache for cancellation flag."""
+            while True:
+                if _task_cache is not None and await _task_cache.get(
+                    f"cancel:{task_id}"
+                ):
+                    return True
+                await asyncio.sleep(0.1)
+
+        async def poll_for_cancel_aiocache() -> bool:
+            """Poll aiocache for cancellation flag."""
+            cache = caches.get("default")
             while True:
                 if await cache.get(f"cancel:{task_id}"):
                     return True
@@ -153,8 +165,14 @@ async def poll_for_cancel() -> bool:
 
         async def watch_for_cancel() -> bool:
             """Watch for cancellation events via pub/sub or polling."""
+            if _task_cache is not None:
+                # ValkeyCache: use polling (pub/sub not implemented yet)
+                return await poll_for_cancel_valkey()
+
+            # aiocache: use pub/sub if Redis, otherwise poll
+            cache = caches.get("default")
             if isinstance(cache, SimpleMemoryCache):
-                return await poll_for_cancel()
+                return await poll_for_cancel_aiocache()
 
             try:
                 client = cache.client
@@ -168,7 +186,7 @@ async def watch_for_cancel() -> bool:
                     "Cancel watcher Redis error, falling back to polling",
                     extra={"task_id": task_id, "error": str(e)},
                 )
-                return await poll_for_cancel()
+                return await poll_for_cancel_aiocache()
             return False
 
         execute_task = asyncio.create_task(fn(*args, **kwargs))
@@ -190,7 +208,12 @@ async def watch_for_cancel() -> bool:
             cancel_watch.cancel()
             return execute_task.result()
         finally:
-            await cache.delete(f"cancel:{task_id}")
+            # Clean up cancellation flag
+            if _task_cache is not None:
+                await _task_cache.delete(f"cancel:{task_id}")
+            else:
+                cache = caches.get("default")
+                await cache.delete(f"cancel:{task_id}")
 
     return wrapper
 
@@ -475,6 +498,8 @@ async def cancel(
     if task_id is None or context_id is None:
         raise ServerError(InvalidParamsError(message="task_id and context_id required"))
 
+    _ensure_task_cache()
+
     if context.current_task and context.current_task.status.state in (
         TaskState.completed,
         TaskState.failed,
@@ -482,11 +507,16 @@ async def cancel(
     ):
         return context.current_task
 
-    cache = caches.get("default")
-
-    await cache.set(f"cancel:{task_id}", True, ttl=3600)
-    if not isinstance(cache, SimpleMemoryCache):
-        await cache.client.publish(f"cancel:{task_id}", "cancel")
+    if _task_cache is not None:
+        # Use ValkeyCache
+        await _task_cache.set(f"cancel:{task_id}", True, ttl=3600)
+        # Note: pub/sub not implemented for ValkeyCache yet, relies on polling
+    else:
+        # Use aiocache
+        cache = caches.get("default")
+        await cache.set(f"cancel:{task_id}", True, ttl=3600)
+        if not isinstance(cache, SimpleMemoryCache):
+            await cache.client.publish(f"cancel:{task_id}", "cancel")
 
     await event_queue.enqueue_event(
         TaskStatusUpdateEvent(
diff --git a/lib/crewai/src/crewai/memory/storage/valkey_cache.py b/lib/crewai/src/crewai/memory/storage/valkey_cache.py
new file mode 100644
index 0000000000..b713655764
--- /dev/null
+++ b/lib/crewai/src/crewai/memory/storage/valkey_cache.py
@@ -0,0 +1,189 @@
+"""Valkey-based cache implementation for CrewAI.
+
+This module provides a simple cache interface using Valkey-GLIDE client
+for caching operations with optional TTL support. It replaces Redis usage
+in A2A communication, file uploads, and agent card caching.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from glide import GlideClient, GlideClientConfiguration, NodeAddress
+
+
+_logger = logging.getLogger(__name__)
+
+
+class ValkeyCache:
+    """Simple cache interface using Valkey-GLIDE client.
+
+    Provides get/set/delete/exists operations for caching with optional TTL.
+    Uses JSON serialization for complex values and lazy client initialization.
+
+    Example:
+        >>> cache = ValkeyCache(host="localhost", port=6379)
+        >>> await cache.set("key", {"data": "value"}, ttl=3600)
+        >>> value = await cache.get("key")
+        >>> await cache.delete("key")
+    """
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 6379,
+        db: int = 0,
+        password: str | None = None,
+        default_ttl: int | None = None,
+    ) -> None:
+        """Initialize Valkey cache.
+
+        Args:
+            host: Valkey server hostname.
+            port: Valkey server port.
+            db: Database number to use.
+            password: Optional password for authentication.
+            default_ttl: Default TTL in seconds (None = no expiration).
+        """
+        self._host = host
+        self._port = port
+        self._db = db
+        self._password = password
+        self._default_ttl = default_ttl
+        self._client: GlideClient | None = None
+
+    async def _get_client(self) -> GlideClient:
+        """Get or create Valkey client (lazy initialization).
+
+        Returns:
+            Initialized GlideClient instance.
+
+        Raises:
+            RuntimeError: If connection to Valkey fails.
+            TimeoutError: If connection attempt times out (10 seconds).
+        """
+        import asyncio
+
+        if self._client is None:
+            host = self._host
+            port = self._port
+            db = self._db
+            try:
+                from glide import ServerCredentials
+
+                config = GlideClientConfiguration(
+                    addresses=[NodeAddress(host, port)],
+                    database_id=db,
+                    credentials=(
+                        ServerCredentials(password=self._password)
+                        if self._password
+                        else None
+                    ),
+                )
+
+                # Add connection timeout (10 seconds)
+                try:
+                    self._client = await asyncio.wait_for(
+                        GlideClient.create(config), timeout=10.0
+                    )
+                except asyncio.TimeoutError as e:
+                    _logger.error("Connection timeout connecting to Valkey")
+                    raise TimeoutError(
+                        "Connection timeout to Valkey. "
+                        "Ensure Valkey is running and accessible."
+                    ) from e
+
+                _logger.info("Valkey cache client initialized")
+            except (TimeoutError, RuntimeError):
+                raise
+            except Exception as e:
+                _logger.error(
+                    "Failed to create Valkey cache client: %s", type(e).__name__
+                )
+                raise RuntimeError(
+                    "Cannot connect to Valkey. Check connection settings."
+                ) from e
+
+        return self._client
+
+    async def get(self, key: str) -> Any | None:
+        """Get value from cache.
+
+        Args:
+            key: Cache key.
+
+        Returns:
+            Cached value (deserialized from JSON) or None if not found.
+        """
+        client = await self._get_client()
+        value = await client.get(key)
+
+        if value is None:
+            return None
+
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            _logger.warning(f"Failed to deserialize cached value for key: {key}")
+            return None
+
+    async def set(
+        self,
+        key: str,
+        value: Any,
+        ttl: int | None = None,
+    ) -> None:
+        """Set value in cache.
+
+        Args:
+            key: Cache key.
+            value: Value to cache (will be serialized to JSON).
+            ttl: TTL in seconds (None uses default_ttl, 0 = no expiration).
+        """
+        from glide import ExpirySet, ExpiryType
+
+        client = await self._get_client()
+        serialized = json.dumps(value)
+
+        ttl_to_use = ttl if ttl is not None else self._default_ttl
+
+        if ttl_to_use and ttl_to_use > 0:
+            # Set with expiration using SET command with EX option
+            await client.set(
+                key,
+                serialized,
+                expiry=ExpirySet(ExpiryType.SEC, ttl_to_use),
+            )
+        else:
+            await client.set(key, serialized)
+
+    async def delete(self, key: str) -> None:
+        """Delete value from cache.
+
+        Args:
+            key: Cache key to delete.
+        """
+        client = await self._get_client()
+        await client.delete([key])
+
+    async def exists(self, key: str) -> bool:
+        """Check if key exists in cache.
+
+        Args:
+            key: Cache key to check.
+
+        Returns:
+            True if key exists, False otherwise.
+        """
+        client = await self._get_client()
+        result = await client.exists([key])
+        return result > 0
+
+    async def close(self) -> None:
+        """Close Valkey client connection."""
+        if self._client:
+            await self._client.close()
+            self._client = None
+            _logger.debug("Valkey cache client closed")
diff --git a/lib/crewai/src/crewai/utilities/cache_config.py b/lib/crewai/src/crewai/utilities/cache_config.py
new file mode 100644
index 0000000000..d13e74383d
--- /dev/null
+++ b/lib/crewai/src/crewai/utilities/cache_config.py
@@ -0,0 +1,66 @@
+"""Shared cache configuration helpers for Valkey/Redis URL parsing."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+from urllib.parse import urlparse
+
+
+_logger = logging.getLogger(__name__)
+
+
+def parse_cache_url() -> dict[str, Any] | None:
+    """Parse VALKEY_URL or REDIS_URL from environment.
+
+    Priority: VALKEY_URL > REDIS_URL.
+
+    Returns:
+        Dict with host, port, db, password keys, or None if no URL is set.
+    """
+    url = os.environ.get("VALKEY_URL") or os.environ.get("REDIS_URL")
+    if not url:
+        return None
+    parsed = urlparse(url)
+    return {
+        "host": parsed.hostname or "localhost",
+        "port": parsed.port or 6379,
+        "db": (
+            int(parsed.path.lstrip("/")) if parsed.path and parsed.path != "/" else 0
+        ),
+        "password": parsed.password,
+    }
+
+
+def get_aiocache_config() -> dict[str, Any]:
+    """Build an aiocache configuration dict from environment.
+
+    Uses VALKEY_URL or REDIS_URL (both are Redis-wire-compatible) to
+    configure ``aiocache.RedisCache``.  Falls back to
+    ``aiocache.SimpleMemoryCache`` when neither variable is set.
+
+    Returns:
+        Configuration dict suitable for ``aiocache.caches.set_config()``.
+    """
+    conn = parse_cache_url()
+    if conn is not None:
+        return {
+            "default": {
+                "cache": "aiocache.RedisCache",
+                "endpoint": conn["host"],
+                "port": conn["port"],
+                "db": conn.get("db", 0),
+                "password": conn.get("password"),
+            }
+        }
+    return {
+        "default": {
+            "cache": "aiocache.SimpleMemoryCache",
+        }
+    }
+
+
+def use_valkey_cache() -> bool:
+    """Return True if VALKEY_URL is set in the environment."""
+    return bool(os.environ.get("VALKEY_URL"))
diff --git a/lib/crewai/tests/memory/storage/test_valkey_cache.py b/lib/crewai/tests/memory/storage/test_valkey_cache.py
new file mode 100644
index 0000000000..7534368aa4
--- /dev/null
+++ b/lib/crewai/tests/memory/storage/test_valkey_cache.py
@@ -0,0 +1,498 @@
+"""Tests for ValkeyCache implementation."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from crewai.memory.storage.valkey_cache import ValkeyCache
+
+
+@pytest.fixture
+def mock_glide_client() -> AsyncMock:
+    """Create a mock GlideClient for testing."""
+    client = AsyncMock()
+    client.get = AsyncMock()
+    client.set = AsyncMock()
+    client.delete = AsyncMock()
+    client.exists = AsyncMock()
+    client.close = AsyncMock()
+    return client
+
+
+@pytest.fixture
+def valkey_cache(mock_glide_client: AsyncMock) -> ValkeyCache:
+    """Create a ValkeyCache instance with mocked client."""
+    cache = ValkeyCache(host="localhost", port=6379, db=0)
+
+    # Mock the client creation to return our mock
+    async def mock_create_client() -> AsyncMock:
+        cache._client = mock_glide_client
+        return mock_glide_client
+
+    cache._get_client = mock_create_client  # type: ignore[method-assign]
+    return cache
+
+
+class TestValkeyCacheBasicOperations:
+    """Tests for basic ValkeyCache operations (get/set/delete/exists)."""
+
+    @pytest.mark.asyncio
+    async def test_set_and_get_string_value(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting and getting a string value."""
+        # Mock get to return serialized string
+        mock_glide_client.get.return_value = json.dumps("test_value")
+
+        # Set value
+        await valkey_cache.set("test_key", "test_value")
+
+        # Verify set was called
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert call_args[0][0] == "test_key"
+        assert call_args[0][1] == json.dumps("test_value")
+
+        # Get value
+        result = await valkey_cache.get("test_key")
+
+        # Verify get was called and result is correct
+        mock_glide_client.get.assert_called_once_with("test_key")
+        assert result == "test_value"
+
+    @pytest.mark.asyncio
+    async def test_set_and_get_dict_value(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting and getting a dictionary value."""
+        test_dict = {"key1": "value1", "key2": 42, "key3": [1, 2, 3]}
+        mock_glide_client.get.return_value = json.dumps(test_dict)
+
+        # Set value
+        await valkey_cache.set("dict_key", test_dict)
+
+        # Verify set was called with serialized dict
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert call_args[0][0] == "dict_key"
+        assert call_args[0][1] == json.dumps(test_dict)
+
+        # Get value
+        result = await valkey_cache.get("dict_key")
+
+        # Verify result matches original dict
+        assert result == test_dict
+
+    @pytest.mark.asyncio
+    async def test_set_and_get_list_value(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting and getting a list value."""
+        test_list = [1, "two", 3.0, {"nested": "dict"}]
+        mock_glide_client.get.return_value = json.dumps(test_list)
+
+        await valkey_cache.set("list_key", test_list)
+        result = await valkey_cache.get("list_key")
+
+        assert result == test_list
+
+    @pytest.mark.asyncio
+    async def test_get_nonexistent_key_returns_none(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test getting a non-existent key returns None."""
+        mock_glide_client.get.return_value = None
+
+        result = await valkey_cache.get("nonexistent_key")
+
+        assert result is None
+        mock_glide_client.get.assert_called_once_with("nonexistent_key")
+
+    @pytest.mark.asyncio
+    async def test_delete_key(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test deleting a key."""
+        await valkey_cache.delete("test_key")
+
+        mock_glide_client.delete.assert_called_once_with(["test_key"])
+
+    @pytest.mark.asyncio
+    async def test_exists_returns_true_for_existing_key(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test exists returns True for existing key."""
+        mock_glide_client.exists.return_value = 1
+
+        result = await valkey_cache.exists("existing_key")
+
+        assert result is True
+        mock_glide_client.exists.assert_called_once_with(["existing_key"])
+
+    @pytest.mark.asyncio
+    async def test_exists_returns_false_for_nonexistent_key(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test exists returns False for non-existent key."""
+        mock_glide_client.exists.return_value = 0
+
+        result = await valkey_cache.exists("nonexistent_key")
+
+        assert result is False
+        mock_glide_client.exists.assert_called_once_with(["nonexistent_key"])
+
+
+class TestValkeyCacheTTL:
+    """Tests for ValkeyCache TTL functionality."""
+
+    @pytest.mark.asyncio
+    async def test_set_with_explicit_ttl(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting a value with explicit TTL."""
+        await valkey_cache.set("ttl_key", "value", ttl=3600)
+
+        # Verify set was called with expiry
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert call_args[0][0] == "ttl_key"
+        assert call_args[0][1] == json.dumps("value")
+        assert "expiry" in call_args[1]
+
+    @pytest.mark.asyncio
+    async def test_set_with_default_ttl(
+        self, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting a value with default TTL from constructor."""
+        cache = ValkeyCache(host="localhost", port=6379, default_ttl=1800)
+
+        async def mock_create_client() -> AsyncMock:
+            cache._client = mock_glide_client
+            return mock_glide_client
+
+        cache._get_client = mock_create_client  # type: ignore[method-assign]
+
+        await cache.set("default_ttl_key", "value")
+
+        # Verify set was called with default TTL
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert "expiry" in call_args[1]
+
+    @pytest.mark.asyncio
+    async def test_set_without_ttl(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting a value without TTL (no expiration)."""
+        await valkey_cache.set("no_ttl_key", "value")
+
+        # Verify set was called without expiry
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert call_args[0][0] == "no_ttl_key"
+        assert call_args[0][1] == json.dumps("value")
+        # Should not have expiry parameter
+        assert "expiry" not in call_args[1] or call_args[1].get("expiry") is None
+
+    @pytest.mark.asyncio
+    async def test_set_with_zero_ttl_no_expiration(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting a value with TTL=0 means no expiration."""
+        await valkey_cache.set("zero_ttl_key", "value", ttl=0)
+
+        # Verify set was called without expiry
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert "expiry" not in call_args[1] or call_args[1].get("expiry") is None
+
+    @pytest.mark.asyncio
+    async def test_explicit_ttl_overrides_default(
+        self, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test explicit TTL overrides default TTL."""
+        cache = ValkeyCache(host="localhost", port=6379, default_ttl=1800)
+
+        async def mock_create_client() -> AsyncMock:
+            cache._client = mock_glide_client
+            return mock_glide_client
+
+        cache._get_client = mock_create_client  # type: ignore[method-assign]
+
+        await cache.set("override_key", "value", ttl=7200)
+
+        # Verify set was called with explicit TTL (7200), not default (1800)
+        mock_glide_client.set.assert_called_once()
+        call_args = mock_glide_client.set.call_args
+        assert "expiry" in call_args[1]
+
+
+class TestValkeyCacheJSONSerialization:
+    """Tests for ValkeyCache JSON serialization edge cases."""
+
+    @pytest.mark.asyncio
+    async def test_serialize_none_value(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing None value."""
+        mock_glide_client.get.return_value = json.dumps(None)
+
+        await valkey_cache.set("none_key", None)
+        result = await valkey_cache.get("none_key")
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_serialize_boolean_values(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing boolean values."""
+        mock_glide_client.get.side_effect = [
+            json.dumps(True),
+            json.dumps(False),
+        ]
+
+        await valkey_cache.set("true_key", True)
+        await valkey_cache.set("false_key", False)
+
+        result_true = await valkey_cache.get("true_key")
+        result_false = await valkey_cache.get("false_key")
+
+        assert result_true is True
+        assert result_false is False
+
+    @pytest.mark.asyncio
+    async def test_serialize_numeric_values(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing numeric values (int, float)."""
+        mock_glide_client.get.side_effect = [
+            json.dumps(42),
+            json.dumps(3.14159),
+            json.dumps(0),
+            json.dumps(-100),
+        ]
+
+        await valkey_cache.set("int_key", 42)
+        await valkey_cache.set("float_key", 3.14159)
+        await valkey_cache.set("zero_key", 0)
+        await valkey_cache.set("negative_key", -100)
+
+        assert await valkey_cache.get("int_key") == 42
+        assert await valkey_cache.get("float_key") == 3.14159
+        assert await valkey_cache.get("zero_key") == 0
+        assert await valkey_cache.get("negative_key") == -100
+
+    @pytest.mark.asyncio
+    async def test_serialize_empty_collections(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing empty collections."""
+        mock_glide_client.get.side_effect = [
+            json.dumps([]),
+            json.dumps({}),
+            json.dumps(""),
+        ]
+
+        await valkey_cache.set("empty_list", [])
+        await valkey_cache.set("empty_dict", {})
+        await valkey_cache.set("empty_string", "")
+
+        assert await valkey_cache.get("empty_list") == []
+        assert await valkey_cache.get("empty_dict") == {}
+        assert await valkey_cache.get("empty_string") == ""
+
+    @pytest.mark.asyncio
+    async def test_serialize_nested_structures(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing deeply nested structures."""
+        nested_data = {
+            "level1": {
+                "level2": {
+                    "level3": [1, 2, {"level4": "deep"}]
+                }
+            },
+            "list": [{"a": 1}, {"b": 2}]
+        }
+        mock_glide_client.get.return_value = json.dumps(nested_data)
+
+        await valkey_cache.set("nested_key", nested_data)
+        result = await valkey_cache.get("nested_key")
+
+        assert result == nested_data
+
+    @pytest.mark.asyncio
+    async def test_deserialize_invalid_json_returns_none(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test deserializing invalid JSON returns None and logs warning."""
+        mock_glide_client.get.return_value = "invalid json {{"
+
+        with patch("crewai.memory.storage.valkey_cache._logger") as mock_logger:
+            result = await valkey_cache.get("invalid_key")
+
+            assert result is None
+            mock_logger.warning.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_serialize_unicode_strings(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing unicode strings."""
+        unicode_data = "Hello 世界 🌍 Привет"
+        mock_glide_client.get.return_value = json.dumps(unicode_data)
+
+        await valkey_cache.set("unicode_key", unicode_data)
+        result = await valkey_cache.get("unicode_key")
+
+        assert result == unicode_data
+
+
+class TestValkeyCacheConnectionManagement:
+    """Tests for ValkeyCache connection management."""
+
+    @pytest.mark.asyncio
+    async def test_lazy_client_initialization(self) -> None:
+        """Test client is initialized lazily on first use."""
+        cache = ValkeyCache(host="localhost", port=6379)
+
+        # Client should be None initially
+        assert cache._client is None
+
+        # Mock GlideClient.create
+        with patch("crewai.memory.storage.valkey_cache.GlideClient") as mock_glide:
+            mock_client = AsyncMock()
+            mock_glide.create = AsyncMock(return_value=mock_client)
+            mock_client.get = AsyncMock(return_value=None)
+
+            # First operation should initialize client
+            await cache.get("test_key")
+
+            # Client should now be initialized
+            assert cache._client is not None
+            mock_glide.create.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_client_reuse_across_operations(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test client is reused across multiple operations."""
+        mock_glide_client.get.return_value = json.dumps("value")
+        mock_glide_client.exists.return_value = 1
+
+        # Perform multiple operations
+        await valkey_cache.get("key1")
+        await valkey_cache.set("key2", "value2")
+        await valkey_cache.exists("key3")
+        await valkey_cache.delete("key4")
+
+        # _get_client should return the same client instance
+        client1 = await valkey_cache._get_client()
+        client2 = await valkey_cache._get_client()
+        assert client1 is client2
+
+    @pytest.mark.asyncio
+    async def test_close_connection(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test closing the client connection."""
+        # Initialize client
+        await valkey_cache._get_client()
+        assert valkey_cache._client is not None
+
+        # Close connection
+        await valkey_cache.close()
+
+        # Verify close was called and client is None
+        mock_glide_client.close.assert_called_once()
+        assert valkey_cache._client is None
+
+    @pytest.mark.asyncio
+    async def test_connection_error_raises_runtime_error(self) -> None:
+        """Test connection error raises RuntimeError with descriptive message."""
+        cache = ValkeyCache(host="invalid-host", port=9999)
+
+        with patch("crewai.memory.storage.valkey_cache.GlideClient") as mock_glide:
+            mock_glide.create = AsyncMock(side_effect=Exception("Connection refused"))
+
+            with pytest.raises(RuntimeError) as exc_info:
+                await cache._get_client()
+
+            assert "Cannot connect to Valkey" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_authentication_with_password(self) -> None:
+        """Test client initialization with password authentication."""
+        cache = ValkeyCache(
+            host="localhost",
+            port=6379,
+            password="secret_password"
+        )
+
+        with patch("crewai.memory.storage.valkey_cache.GlideClient") as mock_glide:
+            mock_client = AsyncMock()
+            mock_glide.create = AsyncMock(return_value=mock_client)
+
+            await cache._get_client()
+
+            # Verify GlideClient.create was called with credentials
+            mock_glide.create.assert_called_once()
+            config = mock_glide.create.call_args[0][0]
+            assert hasattr(config, "credentials")
+
+
+class TestValkeyCacheEdgeCases:
+    """Tests for ValkeyCache edge cases and error conditions."""
+
+    @pytest.mark.asyncio
+    async def test_set_with_special_characters_in_key(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test setting values with special characters in key."""
+        special_keys = [
+            "key:with:colons",
+            "key/with/slashes",
+            "key-with-dashes",
+            "key_with_underscores",
+            "key.with.dots",
+        ]
+
+        for key in special_keys:
+            await valkey_cache.set(key, "value")
+            mock_glide_client.set.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_large_value_serialization(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test serializing large values."""
+        large_list = list(range(10000))
+        mock_glide_client.get.return_value = json.dumps(large_list)
+
+        await valkey_cache.set("large_key", large_list)
+        result = await valkey_cache.get("large_key")
+
+        assert result == large_list
+
+    @pytest.mark.asyncio
+    async def test_concurrent_operations(
+        self, valkey_cache: ValkeyCache, mock_glide_client: AsyncMock
+    ) -> None:
+        """Test concurrent cache operations."""
+        import asyncio
+
+        mock_glide_client.get.return_value = json.dumps("value")
+
+        # Perform concurrent operations
+        tasks = [
+            valkey_cache.set(f"key{i}", f"value{i}")
+            for i in range(10)
+        ]
+        await asyncio.gather(*tasks)
+
+        # Verify all operations completed
+        assert mock_glide_client.set.call_count == 10
diff --git a/lib/crewai/tests/utilities/test_cache_config.py b/lib/crewai/tests/utilities/test_cache_config.py
new file mode 100644
index 0000000000..de28b52f32
--- /dev/null
+++ b/lib/crewai/tests/utilities/test_cache_config.py
@@ -0,0 +1,117 @@
+"""Tests for shared cache configuration helpers."""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import pytest
+
+from crewai.utilities.cache_config import (
+    get_aiocache_config,
+    parse_cache_url,
+    use_valkey_cache,
+)
+
+
+class TestParseCacheUrl:
+    """Tests for parse_cache_url()."""
+
+    def test_returns_none_when_no_env_vars(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            assert parse_cache_url() is None
+
+    def test_parses_valkey_url(self) -> None:
+        with patch.dict(
+            os.environ, {"VALKEY_URL": "redis://myhost:6380/2"}, clear=True
+        ):
+            result = parse_cache_url()
+            assert result is not None
+            assert result["host"] == "myhost"
+            assert result["port"] == 6380
+            assert result["db"] == 2
+            assert result["password"] is None
+
+    def test_parses_redis_url(self) -> None:
+        with patch.dict(
+            os.environ, {"REDIS_URL": "redis://localhost:6379/0"}, clear=True
+        ):
+            result = parse_cache_url()
+            assert result is not None
+            assert result["host"] == "localhost"
+            assert result["port"] == 6379
+            assert result["db"] == 0
+
+    def test_valkey_url_takes_priority_over_redis_url(self) -> None:
+        with patch.dict(
+            os.environ,
+            {
+                "VALKEY_URL": "redis://valkey-host:6380/1",
+                "REDIS_URL": "redis://redis-host:6379/0",
+            },
+            clear=True,
+        ):
+            result = parse_cache_url()
+            assert result is not None
+            assert result["host"] == "valkey-host"
+            assert result["port"] == 6380
+
+    def test_parses_password(self) -> None:
+        with patch.dict(
+            os.environ,
+            {"VALKEY_URL": "redis://:s3cret@myhost:6379/0"},
+            clear=True,
+        ):
+            result = parse_cache_url()
+            assert result is not None
+            assert result["password"] == "s3cret"
+
+    def test_defaults_for_minimal_url(self) -> None:
+        with patch.dict(
+            os.environ, {"VALKEY_URL": "redis://myhost"}, clear=True
+        ):
+            result = parse_cache_url()
+            assert result is not None
+            assert result["host"] == "myhost"
+            assert result["port"] == 6379
+            assert result["db"] == 0
+            assert result["password"] is None
+
+
+class TestGetAiocacheConfig:
+    """Tests for get_aiocache_config()."""
+
+    def test_returns_memory_cache_when_no_url(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            config = get_aiocache_config()
+            assert config["default"]["cache"] == "aiocache.SimpleMemoryCache"
+
+    def test_returns_redis_cache_when_url_set(self) -> None:
+        with patch.dict(
+            os.environ, {"VALKEY_URL": "redis://myhost:6380/2"}, clear=True
+        ):
+            config = get_aiocache_config()
+            assert config["default"]["cache"] == "aiocache.RedisCache"
+            assert config["default"]["endpoint"] == "myhost"
+            assert config["default"]["port"] == 6380
+            assert config["default"]["db"] == 2
+
+
+class TestUseValkeyCache:
+    """Tests for use_valkey_cache()."""
+
+    def test_returns_false_when_not_set(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            assert use_valkey_cache() is False
+
+    def test_returns_true_when_set(self) -> None:
+        with patch.dict(
+            os.environ, {"VALKEY_URL": "redis://localhost:6379"}, clear=True
+        ):
+            assert use_valkey_cache() is True
+
+    def test_returns_false_when_only_redis_url_set(self) -> None:
+        with patch.dict(
+            os.environ, {"REDIS_URL": "redis://localhost:6379"}, clear=True
+        ):
+            assert use_valkey_cache() is False
diff --git a/uv.lock b/uv.lock
index 6e5f936532..2e1e340858 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1365,6 +1365,9 @@ qdrant-edge = [
 tools = [
     { name = "crewai-tools" },
 ]
+valkey = [
+    { name = "valkey-glide" },
+]
 voyageai = [
     { name = "voyageai" },
 ]
@@ -1428,7 +1431,7 @@ requires-dist = [
     { name = "tomli-w", specifier = "~=1.1.0" },
     { name = "voyageai", marker = "extra == 'voyageai'", specifier = "~=0.3.5" },
 ]
-provides-extras = ["a2a", "anthropic", "aws", "azure-ai-inference", "bedrock", "docling", "embeddings", "file-processing", "google-genai", "litellm", "mem0", "openpyxl", "pandas", "qdrant", "qdrant-edge", "tools", "voyageai", "watson"]
+provides-extras = ["a2a", "anthropic", "aws", "azure-ai-inference", "bedrock", "docling", "embeddings", "file-processing", "google-genai", "litellm", "mem0", "openpyxl", "pandas", "qdrant", "qdrant-edge", "tools", "valkey", "voyageai", "watson"]
 
 [[package]]
 name = "crewai-cli"
@@ -9520,6 +9523,43 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/fa/6e/3e955517e22cbdd565f2f8b2e73d52528b14b8bcfdb04f62466b071de847/validators-0.35.0-py3-none-any.whl", hash = "sha256:e8c947097eae7892cb3d26868d637f79f47b4a0554bc6b80065dfe5aac3705dd", size = 44712, upload-time = "2025-05-01T05:42:04.203Z" },
 ]
 
+[[package]]
+name = "valkey-glide"
+version = "2.0.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "protobuf" },
+    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/32/35/fb0401c4bc7be748d937e95213786d21d9e56767b3ad816db5bad6f92c01/valkey_glide-2.0.1.tar.gz", hash = "sha256:4f9c62a88aedffd725cced7d28a9488b27e3f675d1a5294b4962624e97d346c4", size = 1026255, upload-time = "2025-06-20T01:08:15.861Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/44/a3/bf5ff3841538d0bb337371e073dc2c0e93f748f7f8b10a44806f36ab5fa1/valkey_glide-2.0.1-cp310-cp310-macosx_10_7_x86_64.whl", hash = "sha256:b3307934b76557b18ac559f327592cc09fc895fc653ba46010dd6d70fb6239dc", size = 5074638, upload-time = "2025-06-20T01:07:30.16Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/c4/20b66dced96bdca81aa294b39bc03018ed22628c52076752e8d1d3540a7d/valkey_glide-2.0.1-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:6b83d34e2e723e97c41682479b0dce5882069066e808316292b363855992b449", size = 4750261, upload-time = "2025-06-20T01:07:32.452Z" },
+    { url = "https://files.pythonhosted.org/packages/53/58/6440e66bde8963d86bc3c44d88f993059f2a9d7ebdb3256a695d035cff50/valkey_glide-2.0.1-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:1baaf14d09d464ae645be5bdb5dc6b8a38b7eacf22f9dcb2907200c74fbdcdd3", size = 4767755, upload-time = "2025-06-20T01:07:33.86Z" },
+    { url = "https://files.pythonhosted.org/packages/3b/69/dd5c350ce4d2cadde0d83beb601f05e1e62622895f268135e252e8bfc307/valkey_glide-2.0.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4427e7b4d54c9de289a35032c19d5956f94376f5d4335206c5ac4524cbd1c64a", size = 5094507, upload-time = "2025-06-20T01:07:35.349Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/dd/0dd6614e09123a5bd7273bf1159c958d1ea65e7decc2190b225d212e0cb9/valkey_glide-2.0.1-cp311-cp311-macosx_10_7_x86_64.whl", hash = "sha256:6379582d6fbd817697fb119274e37d397db450103cd15d4bd71e555e6d88fb6b", size = 5072939, upload-time = "2025-06-20T01:07:36.948Z" },
+    { url = "https://files.pythonhosted.org/packages/c6/04/986188e407231a5f0bfaf31f31b68e3605ab66f4f4c656adfbb0345669d9/valkey_glide-2.0.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:0f1c0fe003026d8ae172369e0eb2337cbff16f41d4c085332487d6ca2e5282e6", size = 4750491, upload-time = "2025-06-20T01:07:38.659Z" },
+    { url = "https://files.pythonhosted.org/packages/ac/fb/2f5cec71ae51c464502a892b6825426cd74a2c325827981726e557926c94/valkey_glide-2.0.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:82c5f33598e50bcfec6fc924864931f3c6e30cd327a9c9562e1c7ac4e17e79fd", size = 4767597, upload-time = "2025-06-20T01:07:40.091Z" },
+    { url = "https://files.pythonhosted.org/packages/3a/31/851a1a734fe5da5d520106fcfd824e4da09c3be8a0a2123bb4b1980db1ea/valkey_glide-2.0.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:79039a9dc23bb074680f171c12b36b3322357a0af85125534993e81a619dce21", size = 5094383, upload-time = "2025-06-20T01:07:41.329Z" },
+    { url = "https://files.pythonhosted.org/packages/fc/6d/1e7b432cbc02fe63e7496b984b7fc830fb7de388c877b237e0579a6300fc/valkey_glide-2.0.1-cp312-cp312-macosx_10_7_x86_64.whl", hash = "sha256:f55ec8968b0fde364a5b3399be34b89dcb9068994b5cd384e20db0773ad12723", size = 5075024, upload-time = "2025-06-20T01:07:42.917Z" },
+    { url = "https://files.pythonhosted.org/packages/ca/39/6e9f83970590d17d19f596e1b3a366d39077624888e3dd709309efc67690/valkey_glide-2.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:21598f49313912ad27dc700d7b13a3b4bfed7ed9dffad207235cac7d218f4966", size = 4748418, upload-time = "2025-06-20T01:07:44.64Z" },
+    { url = "https://files.pythonhosted.org/packages/98/0e/91335c13dc8e7ceb95063234c16010b46e2dd874a2edef62dea155081647/valkey_glide-2.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f662285146529328e2b5a0a7047f699339b4e0d250eb1f252b15c9befa0dea05", size = 4767264, upload-time = "2025-06-20T01:07:46.185Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/94/ee4d9d441f83fec1464d9f4e52f7940bdd2aeb917589e6abd57498880876/valkey_glide-2.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3939aaa8411fcbba00cb1ff7c7ba73f388bb1deca919972f65cba7eda1d5fa95", size = 5093543, upload-time = "2025-06-20T01:07:47.345Z" },
+    { url = "https://files.pythonhosted.org/packages/ed/7e/257a2e4b61ac29d5923f89bad5fe62be7b4a19e7bec78d191af3ce77aa39/valkey_glide-2.0.1-cp313-cp313-macosx_10_7_x86_64.whl", hash = "sha256:c49b53011a05b5820d0c660ee5c76574183b413a54faa33cf5c01ce77164d9c8", size = 5073114, upload-time = "2025-06-20T01:07:48.885Z" },
+    { url = "https://files.pythonhosted.org/packages/20/14/a8a470679953980af7eac3ccb09638f2a76d4547116d48cbc69ae6f25080/valkey_glide-2.0.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:3a23572b83877537916ba36ad0a6b2fd96581534f0bc67ef8f8498bf4dbb2b40", size = 4747717, upload-time = "2025-06-20T01:07:50.092Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/49/f168dd0c778d9f6ff1be70d5d3bad7a86928fee563de7de5f4f575eddfd8/valkey_glide-2.0.1-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:943a2c4a5c38b8a6b53281201d5a4997ec454a6fdda72d27050eeb6aaef12afb", size = 4767128, upload-time = "2025-06-20T01:07:51.306Z" },
+    { url = "https://files.pythonhosted.org/packages/43/be/68961b14ea133d1792ce50f6df1753848b5377c3e06a8dbe4e39188a549a/valkey_glide-2.0.1-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d770ec581acc59d5597e7ccaac37aee7e3b5e716a77a7fa44e2967db3a715f53", size = 5093522, upload-time = "2025-06-20T01:07:52.546Z" },
+    { url = "https://files.pythonhosted.org/packages/51/2e/ad8595ffe84317385d52ceab8de1e9ef06a4da6b81ca8cd61b7961923de4/valkey_glide-2.0.1-pp310-pypy310_pp73-macosx_10_7_x86_64.whl", hash = "sha256:d4a9ccfe2b190c90622849dab62f9468acf76a282719a1245d272b649e7c12d1", size = 5074539, upload-time = "2025-06-20T01:07:59.87Z" },
+    { url = "https://files.pythonhosted.org/packages/db/e5/2122541c7a64706f3631655209bb0b13723fb99db3c190d9a792b4e7d494/valkey_glide-2.0.1-pp310-pypy310_pp73-macosx_11_0_arm64.whl", hash = "sha256:9aa004077b82f64b23ea0d38d948b5116c23f7228dae3a5b4fcfa1799f8ff7de", size = 4753222, upload-time = "2025-06-20T01:08:01.376Z" },
+    { url = "https://files.pythonhosted.org/packages/6c/13/cd9a20988a820ff61b127d3f850887b28bb734daf2c26d512d8e4c2e8e9e/valkey_glide-2.0.1-pp310-pypy310_pp73-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:631a7a0e2045f7e5e3706e1903beeddf381a6529e318c27230798f4382579e4f", size = 4771530, upload-time = "2025-06-20T01:08:02.6Z" },
+    { url = "https://files.pythonhosted.org/packages/c7/fc/047e89cc01b4cc71db1b6b8160d3b5d050097b408028022c002351238641/valkey_glide-2.0.1-pp310-pypy310_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3c5ed905fb62368c9bc6aef9df8d66269ef51f968dc527da4d7c956927382c1d", size = 5091242, upload-time = "2025-06-20T01:08:04.111Z" },
+    { url = "https://files.pythonhosted.org/packages/1c/9e/68790c1a263f3a0094d67d0109be34631f6f79c2fbce5ced7e33a65ad363/valkey_glide-2.0.1-pp311-pypy311_pp73-macosx_10_7_x86_64.whl", hash = "sha256:53da3cc47c8d946ac76ecc4b468a469d3486778833a59162ea69aa7ce70cbb27", size = 5072793, upload-time = "2025-06-20T01:08:05.562Z" },
+    { url = "https://files.pythonhosted.org/packages/1f/ae/a935af65ae4069d76c69f28f6bfb4533da8b89f7fc418beb7a1482cdd9ee/valkey_glide-2.0.1-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:e526a7d718cdd299d6b03091c12dcc15cd02ff22fe420f253341a4891c50824d", size = 4753435, upload-time = "2025-06-20T01:08:07.149Z" },
+    { url = "https://files.pythonhosted.org/packages/3b/c2/c91d753a89dd87dce2fc8932cfbe174c7a1226c657b3cd64c063f21d4fe6/valkey_glide-2.0.1-pp311-pypy311_pp73-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:8d3345ea2adf6f745733fa5157d8709bcf5ffbb2674391aeebd8f166a37cbc96", size = 4771401, upload-time = "2025-06-20T01:08:08.359Z" },
+    { url = "https://files.pythonhosted.org/packages/00/fe/ad83cfc2ac87bf6bad2b75fa64fca5a6dd54568c1de551d36d369e07f948/valkey_glide-2.0.1-pp311-pypy311_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d1c5fff0f12d2aa4277ddc335035b2c8e12bb11243c1a0f3c35071f4a8b11064", size = 5091360, upload-time = "2025-06-20T01:08:09.622Z" },
+]
+
 [[package]]
 name = "vcrpy"
 version = "7.0.0"