diff --git a/src/fastmcp/resources/template.py b/src/fastmcp/resources/template.py index cde1d273c..1c978eed6 100644 --- a/src/fastmcp/resources/template.py +++ b/src/fastmcp/resources/template.py @@ -184,10 +184,14 @@ async def resource_read_fn() -> str | bytes: fn=resource_read_fn, uri=uri, name=self.name, + title=self.title, description=self.description, + icons=self.icons, mime_type=self.mime_type, tags=self.tags, enabled=self.enabled, + annotations=self.annotations, + meta=self.meta, ) def to_mcp_template( diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index 5f3931dda..61f74ab70 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -464,9 +464,64 @@ def _setup_handlers(self) -> None: self._mcp_server.call_tool(validate_input=self.strict_input_validation)( self._call_tool_mcp ) - self._mcp_server.read_resource()(self._read_resource_mcp) + # Custom read_resource handler to support _meta in content + self._setup_read_resource_handler() self._mcp_server.get_prompt()(self._get_prompt_mcp) + def _setup_read_resource_handler(self) -> None: + """Set up custom read_resource handler with _meta support.""" + import base64 + + async def handler(req: mcp.types.ReadResourceRequest): + """Custom handler that preserves resource meta in content.""" + uri = req.params.uri + result = await self._read_resource_mcp(uri) + + # Convert ReadResourceContents to proper MCP types with _meta + contents_list: list[ + mcp.types.TextResourceContents | mcp.types.BlobResourceContents + ] = [] + + # Get the resource to access its meta + try: + resource = await self._resource_manager.get_resource(uri) + resource_meta = resource.get_meta( + include_fastmcp_meta=self.include_fastmcp_meta + ) + except Exception: + resource_meta = None + + for content_item in result: + match content_item.content: + case str() as data: + contents_list.append( + mcp.types.TextResourceContents( + uri=uri, + text=data, + mimeType=content_item.mime_type or "text/plain", + _meta=resource_meta, + ) + ) + case bytes() as data: + contents_list.append( + mcp.types.BlobResourceContents( + uri=uri, + blob=base64.b64encode(data).decode(), + mimeType=content_item.mime_type + or "application/octet-stream", + _meta=resource_meta, + ) + ) + + return mcp.types.ServerResult( + mcp.types.ReadResourceResult( + contents=contents_list, + ) + ) + + # Register the custom handler directly + self._mcp_server.request_handlers[mcp.types.ReadResourceRequest] = handler + async def _apply_middleware( self, context: MiddlewareContext[Any], diff --git a/tests/resources/test_resource_content_meta.py b/tests/resources/test_resource_content_meta.py new file mode 100644 index 000000000..0242186f0 --- /dev/null +++ b/tests/resources/test_resource_content_meta.py @@ -0,0 +1,203 @@ +"""Test that _meta attributes are included in resource content responses.""" + +from pydantic import AnyUrl + +from fastmcp import FastMCP +from fastmcp.client import Client + + +class TestResourceContentMetaRawResponse: + """Test that _meta is present in raw MCP response.""" + + async def test_raw_response_includes_meta(self): + """Test that raw MCP response includes _meta in content.""" + mcp = FastMCP("TestServer") + + @mcp.resource(uri="test://raw", meta={"widgetDomain": "example.com"}) + def raw_resource() -> str: + return "Raw content" + + async with Client(mcp) as client: + # Get the raw MCP response + result = await client.read_resource_mcp(AnyUrl("test://raw")) + + # Verify response structure + assert result.contents is not None + assert len(result.contents) == 1 + + # Check the raw content object + content = result.contents[0] + assert hasattr(content, "meta") + assert content.meta is not None + assert "widgetDomain" in content.meta + assert content.meta["widgetDomain"] == "example.com" + + # Verify serialization includes _meta + content_dict = content.model_dump(by_alias=True) + assert "_meta" in content_dict + assert content_dict["_meta"]["widgetDomain"] == "example.com" + + async def test_raw_response_template_includes_meta(self): + """Test that raw MCP response includes _meta for templates.""" + mcp = FastMCP("TestServer") + + @mcp.resource( + uri="test://item/{item_id}", meta={"widgetDomain": "items.example.com"} + ) + def item_resource(item_id: str) -> str: + return f"Item: {item_id}" + + async with Client(mcp) as client: + # Get the raw MCP response + result = await client.read_resource_mcp(AnyUrl("test://item/456")) + + # Verify response structure + assert result.contents is not None + assert len(result.contents) == 1 + + # Check the raw content object + content = result.contents[0] + assert hasattr(content, "meta") + assert content.meta is not None + assert "widgetDomain" in content.meta + assert content.meta["widgetDomain"] == "items.example.com" + + async def test_raw_response_meta_serialization(self): + """Test that _meta is properly serialized in the wire format.""" + mcp = FastMCP("TestServer") + + meta_data = { + "widgetDomain": "example.com", + "version": "2.0", + "author": "test", + } + + @mcp.resource(uri="test://serialize", meta=meta_data) + def serialize_resource() -> str: + return "Serialized content" + + async with Client(mcp) as client: + # Get the raw MCP response + result = await client.read_resource_mcp(AnyUrl("test://serialize")) + + # Verify the content + content = result.contents[0] + + # Serialize to dict and verify all meta fields are present + content_dict = content.model_dump(by_alias=True) + assert "_meta" in content_dict + + # Verify all custom meta fields + for key, value in meta_data.items(): + assert key in content_dict["_meta"] + assert content_dict["_meta"][key] == value + + +class TestResourceContentMeta: + """Test resource content _meta attribute.""" + + async def test_resource_content_includes_meta(self): + """Test that resource content includes _meta from resource.meta.""" + mcp = FastMCP("TestServer") + + @mcp.resource(uri="test://example", meta={"widgetDomain": "example.com"}) + def example_resource() -> str: + return "Example content" + + async with Client(mcp) as client: + # Read the resource + result = await client.read_resource(AnyUrl("test://example")) + + # Verify we got content + assert len(result) == 1 + content = result[0] + + # Verify the content has text + assert content.text == "Example content" # type: ignore[attr-defined] + + # Verify _meta is included in the content + assert content.meta is not None + assert "widgetDomain" in content.meta + assert content.meta["widgetDomain"] == "example.com" + + async def test_resource_template_content_includes_meta(self): + """Test that resource template content includes _meta from template.meta.""" + mcp = FastMCP("TestServer") + + @mcp.resource( + uri="test://user/{user_id}", meta={"widgetDomain": "users.example.com"} + ) + def user_resource(user_id: str) -> str: + return f"User: {user_id}" + + async with Client(mcp) as client: + # Read the resource via template + result = await client.read_resource(AnyUrl("test://user/123")) + + # Verify we got content + assert len(result) == 1 + content = result[0] + + # Verify the content has text + assert content.text == "User: 123" # type: ignore[attr-defined] + + # Verify _meta is included in the content + assert content.meta is not None + assert "widgetDomain" in content.meta + assert content.meta["widgetDomain"] == "users.example.com" + + async def test_resource_content_without_meta(self): + """Test that resources without meta still work.""" + mcp = FastMCP("TestServer") + + @mcp.resource(uri="test://no-meta") + def no_meta_resource() -> str: + return "Content without meta" + + async with Client(mcp) as client: + # Read the resource + result = await client.read_resource(AnyUrl("test://no-meta")) + + # Verify we got content + assert len(result) == 1 + content = result[0] + + # Verify the content has text + assert content.text == "Content without meta" # type: ignore[attr-defined] + + # Meta might be None or might have fastmcp meta depending on settings + # Just verify this doesn't crash + + async def test_resource_content_meta_with_multiple_fields(self): + """Test resource content with multiple meta fields.""" + mcp = FastMCP("TestServer") + + meta_data = { + "widgetDomain": "example.com", + "version": "1.0", + "category": "documentation", + } + + @mcp.resource(uri="test://multi-meta", meta=meta_data) + def multi_meta_resource() -> str: + return "Content with multiple meta fields" + + async with Client(mcp) as client: + # Read the resource + result = await client.read_resource(AnyUrl("test://multi-meta")) + + # Verify we got content + assert len(result) == 1 + content = result[0] + + # Verify the content has text + assert content.text == "Content with multiple meta fields" # type: ignore[attr-defined] + + # Verify all meta fields are included + assert content.meta is not None + assert "widgetDomain" in content.meta + assert content.meta["widgetDomain"] == "example.com" + assert "version" in content.meta + assert content.meta["version"] == "1.0" + assert "category" in content.meta + assert content.meta["category"] == "documentation"