Preserve component metadata in response cache (#4521)

fastmcp_slim/fastmcp/server/middleware/caching.py

@@ -16,7 +16,7 @@ from key_value.aio.wrappers.statistics.wrapper import (
     KVStoreCollectionStatistics,
 )
 from pydantic import Field
-from typing_extensions import NotRequired, Self, override
+from typing_extensions import NotRequired, Self, TypeVar, override
 
 from fastmcp.prompts.base import Message, Prompt, PromptResult
 from fastmcp.resources.base import Resource, ResourceContent, ResourceResult
@@ -36,6 +36,18 @@ ONE_MB_IN_BYTES = 1024 * 1024
 
 ANONYMOUS_AUTH_KEY = "__anonymous__"
 
+BaseModelT = TypeVar("BaseModelT", bound=FastMCPBaseModel)
+
+
+def _to_base_model(value: FastMCPBaseModel, model_type: type[BaseModelT]) -> BaseModelT:
+    """Validate a component's public base fields without serializing its subclass."""
+    field_values = {
+        name: getattr(value, name)
+        for name, field in model_type.model_fields.items()
+        if not field.exclude
+    }
+    return model_type.model_validate(field_values)
+
 
 class CachableResourceContent(FastMCPBaseModel):
     """A wrapper for ResourceContent that can be cached."""
@@ -314,19 +326,7 @@ class ResponseCachingMiddleware(Middleware):
         tools: Sequence[Tool] = await call_next(context)
 
         # Turn any subclass of Tool into a Tool
-        cachable_tools: list[Tool] = [
-            Tool(
-                name=tool.name,
-                title=tool.title,
-                description=tool.description,
-                parameters=tool.parameters,
-                output_schema=tool.output_schema,
-                annotations=tool.annotations,
-                meta=tool.meta,
-                tags=tool.tags,
-            )
-            for tool in tools
-        ]
+        cachable_tools = [_to_base_model(tool, Tool) for tool in tools]
 
         await self._list_tools_cache.put(
             key=cache_key,
@@ -355,18 +355,8 @@ class ResponseCachingMiddleware(Middleware):
         resources: Sequence[Resource] = await call_next(context)
 
         # Turn any subclass of Resource into a Resource
-        cachable_resources: list[Resource] = [
-            Resource(
-                name=resource.name,
-                title=resource.title,
-                description=resource.description,
-                tags=resource.tags,
-                meta=resource.meta,
-                mime_type=resource.mime_type,
-                annotations=resource.annotations,
-                uri=resource.uri,
-            )
-            for resource in resources
+        cachable_resources = [
+            _to_base_model(resource, Resource) for resource in resources
         ]
 
         await self._list_resources_cache.put(
@@ -396,17 +386,7 @@ class ResponseCachingMiddleware(Middleware):
         prompts: Sequence[Prompt] = await call_next(context)
 
         # Turn any subclass of Prompt into a Prompt
-        cachable_prompts: list[Prompt] = [
-            Prompt(
-                name=prompt.name,
-                title=prompt.title,
-                description=prompt.description,
-                tags=prompt.tags,
-                meta=prompt.meta,
-                arguments=prompt.arguments,
-            )
-            for prompt in prompts
-        ]
+        cachable_prompts = [_to_base_model(prompt, Prompt) for prompt in prompts]
 
         await self._list_prompts_cache.put(
             key=cache_key,

tests/server/middleware/test_caching.py

@@ -42,6 +42,7 @@ from fastmcp.server.middleware.caching import (
 )
 from fastmcp.server.middleware.middleware import CallNext, MiddlewareContext
 from fastmcp.tools.base import Tool, ToolResult
+from fastmcp.utilities.tasks import TaskConfig
 
 TEST_URI = AnyUrl("https://test_uri")
 
@@ -352,6 +353,47 @@ class TestResponseCachingMiddlewareIntegration:
 
             assert pre_tool_list == post_tool_list
 
+    async def test_list_operations_preserve_component_metadata(self):
+        """Base component fields should survive conversion through the cache."""
+        icon = mcp_types.Icon(src="https://example.com/component.png")
+        mcp = FastMCP("MetadataServer")
+        mcp.add_middleware(ResponseCachingMiddleware())
+
+        @mcp.tool(icons=[icon], task=TaskConfig(mode="optional"))
+        async def greet() -> str:
+            return "hello"
+
+        @mcp.resource("resource://metadata", icons=[icon])
+        def metadata_resource() -> str:
+            return "resource"
+
+        @mcp.prompt(icons=[icon])
+        def metadata_prompt() -> str:
+            return "prompt"
+
+        cached_tools = await mcp.list_tools()
+        cached_resources = await mcp.list_resources()
+        cached_prompts = await mcp.list_prompts()
+        assert type(cached_tools[0]) is Tool
+        assert type(cached_resources[0]) is Resource
+        assert type(cached_prompts[0]) is Prompt
+        assert not hasattr(cached_tools[0], "fn")
+        assert not hasattr(cached_resources[0], "fn")
+        assert not hasattr(cached_prompts[0], "fn")
+
+        async with Client(mcp) as client:
+            for _ in range(2):
+                tools = await client.list_tools()
+                resources = await client.list_resources()
+                prompts = await client.list_prompts()
+
+                assert tools[0].icons == [icon]
+                assert tools[0].execution == mcp_types.ToolExecution(
+                    task_support="optional"
+                )
+                assert resources[0].icons == [icon]
+                assert prompts[0].icons == [icon]
+
     async def test_call_tool(
         self,
         caching_server: FastMCP,
@@ -595,13 +637,16 @@ class TestCachingWithImportedServerPrefixes:
     ):
         """Resource URIs should retain prefix after being served from cache."""
         async with Client(parent_with_imported_child) as client:
-            # First call populates cache
-            resources_first = await client.list_resources()
-            resource_uris_first = [str(r.uri) for r in resources_first]
+            with warnings.catch_warnings():
+                warnings.simplefilter("error", UserWarning)
 
-            # Second call should come from cache
-            resources_cached = await client.list_resources()
-            resource_uris_cached = [str(r.uri) for r in resources_cached]
+                # First call populates cache
+                resources_first = await client.list_resources()
+                resource_uris_first = [str(r.uri) for r in resources_first]
+
+                # Second call should come from cache
+                resources_cached = await client.list_resources()
+                resource_uris_cached = [str(r.uri) for r in resources_cached]
 
             # All resources should have prefix in URI path in both calls
             # Resources get path-style prefix: resource://child/path