diff --git a/examples/snippets/servers/lowlevel/resource_contents_direct.py b/examples/snippets/servers/lowlevel/resource_contents_direct.py new file mode 100644 index 000000000..1d3adcd51 --- /dev/null +++ b/examples/snippets/servers/lowlevel/resource_contents_direct.py @@ -0,0 +1,273 @@ +""" +Example showing how to return ResourceContents objects directly from +low-level server resources. + +The main benefit is the ability to include metadata (_meta field) with +your resources, providing additional context about the resource content +such as timestamps, versions, authorship, or any domain-specific metadata. +""" + +import asyncio +from collections.abc import Iterable + +from pydantic import AnyUrl + +import mcp.server.stdio as stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.lowlevel.server import ReadResourceContents + +# Create a server instance +server = Server( + name="LowLevel ResourceContents Example", + version="1.0.0", +) + + +# Example 1: Return TextResourceContents directly +@server.read_resource() +async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | types.BlobResourceContents]: + """Handle resource reading with direct ResourceContents return.""" + uri_str = str(uri) + + if uri_str == "text://readme": + # Return TextResourceContents with document metadata + return [ + types.TextResourceContents( + uri=uri, + text="# README\n\nThis is a sample readme file.", + mimeType="text/markdown", + _meta={ + "title": "Project README", + "author": "Development Team", + "lastModified": "2024-01-15T10:00:00Z", + "version": "2.1.0", + "language": "en", + "license": "MIT", + }, + ) + ] + + elif uri_str == "data://config.json": + # Return JSON data with schema and validation metadata + return [ + types.TextResourceContents( + uri=uri, + text='{\n "version": "1.0.0",\n "debug": false\n}', + mimeType="application/json", + _meta={ + "schema": "https://example.com/schemas/config/v1.0", + "validated": True, + "environment": "production", + "lastValidated": "2024-01-15T14:00:00Z", + "checksum": "sha256:abc123...", + }, + ) + ] + + elif uri_str == "image://icon.png": + # Return binary data with comprehensive image metadata + import base64 + + # This is a 1x1 transparent PNG + png_data = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" + ) + return [ + types.BlobResourceContents( + uri=uri, + blob=base64.b64encode(png_data).decode(), + mimeType="image/png", + _meta={ + "width": 1, + "height": 1, + "bitDepth": 8, + "colorType": 6, # RGBA + "compression": 0, + "filter": 0, + "interlace": 0, + "fileSize": len(png_data), + "hasAlpha": True, + "generated": "2024-01-15T12:00:00Z", + "generator": "Example MCP Server", + }, + ) + ] + + elif uri_str == "multi://content": + # Return multiple ResourceContents objects with part metadata + return [ + types.TextResourceContents( + uri=uri, + text="Part 1: Introduction", + mimeType="text/plain", + _meta={ + "part": 1, + "title": "Introduction", + "order": 1, + "required": True, + }, + ), + types.TextResourceContents( + uri=uri, + text="## Part 2: Main Content\n\nThis is the main section.", + mimeType="text/markdown", + _meta={ + "part": 2, + "title": "Main Content", + "order": 2, + "wordCount": 8, + "headingLevel": 2, + }, + ), + types.BlobResourceContents( + uri=uri, + blob="UGFydCAzOiBCaW5hcnkgRGF0YQ==", # "Part 3: Binary Data" in base64 + mimeType="application/octet-stream", + _meta={ + "part": 3, + "title": "Binary Attachment", + "order": 3, + "encoding": "base64", + "originalSize": 19, + }, + ), + ] + + elif uri_str.startswith("code://"): + # Extract language from URI for syntax highlighting + language = uri_str.split("://")[1].split("/")[0] + code_samples = { + "python": ('def hello():\n print("Hello, World!")', "text/x-python"), + "javascript": ('console.log("Hello, World!");', "text/javascript"), + "html": ("

Hello, World!

", "text/html"), + } + + if language in code_samples: + code, mime_type = code_samples[language] + return [ + types.TextResourceContents( + uri=uri, + text=code, + mimeType=mime_type, + _meta={ + "language": language, + "syntaxHighlighting": True, + "lineNumbers": True, + "executable": language in ["python", "javascript"], + "documentation": f"https://docs.example.com/languages/{language}", + }, + ) + ] + + # Default case - resource not found + return [ + types.TextResourceContents( + uri=uri, + text=f"Resource not found: {uri}", + mimeType="text/plain", + ) + ] + + +# List available resources +@server.list_resources() +async def list_resources() -> list[types.Resource]: + """List all available resources.""" + return [ + types.Resource( + uri=AnyUrl("text://readme"), + name="README", + title="README file", + description="A sample readme in markdown format", + mimeType="text/markdown", + ), + types.Resource( + uri=AnyUrl("data://config.json"), + name="config", + title="Configuration", + description="Application configuration in JSON format", + mimeType="application/json", + ), + types.Resource( + uri=AnyUrl("image://icon.png"), + name="icon", + title="Application Icon", + description="A sample PNG icon", + mimeType="image/png", + ), + types.Resource( + uri=AnyUrl("multi://content"), + name="multi-part", + title="Multi-part Content", + description="A resource that returns multiple content items", + mimeType="multipart/mixed", + ), + types.Resource( + uri=AnyUrl("code://python/example"), + name="python-code", + title="Python Code Example", + description="Sample Python code with proper MIME type", + mimeType="text/x-python", + ), + ] + + +# Also demonstrate with ReadResourceContents (old style) mixed in +@server.list_resources() +async def list_legacy_resources() -> list[types.Resource]: + """List resources that use the legacy ReadResourceContents approach.""" + return [ + types.Resource( + uri=AnyUrl("legacy://text"), + name="legacy-text", + title="Legacy Text Resource", + description="Uses ReadResourceContents wrapper", + mimeType="text/plain", + ), + ] + + +# Mix old and new styles to show compatibility +@server.read_resource() +async def read_legacy_resource( + uri: AnyUrl, +) -> Iterable[ReadResourceContents | types.TextResourceContents | types.BlobResourceContents]: + """Handle legacy resources alongside new ResourceContents.""" + uri_str = str(uri) + + if uri_str == "legacy://text": + # Old style - return ReadResourceContents + return [ + ReadResourceContents( + content="This uses the legacy ReadResourceContents wrapper", + mime_type="text/plain", + ) + ] + + # For other resources, return a simple not found message + return [ + types.TextResourceContents( + uri=uri, + text=f"Resource not found: {uri}", + mimeType="text/plain", + ) + ] + + +async def main(): + """Run the server using stdio transport.""" + async with stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + server.create_initialization_options( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ) + + +if __name__ == "__main__": + # Run with: python resource_contents_direct.py + asyncio.run(main()) diff --git a/examples/snippets/servers/resource_contents_direct.py b/examples/snippets/servers/resource_contents_direct.py new file mode 100644 index 000000000..ca569aca4 --- /dev/null +++ b/examples/snippets/servers/resource_contents_direct.py @@ -0,0 +1,190 @@ +""" +Example showing how to return ResourceContents objects directly from resources. + +The main benefit of returning ResourceContents directly is the ability to include +metadata through the _meta field. This allows +you to attach additional context to your resources such as: + +- Timestamps (created, modified, expires) +- Version information +- Author/ownership details +- File system metadata (permissions, size) +- Image metadata (dimensions, color space) +- Document metadata (word count, language) +- Validation status and schemas +- Any domain-specific metadata + +This metadata helps clients better understand and work with the resource content. +""" + +import base64 + +from pydantic import AnyUrl + +from mcp.server.fastmcp import FastMCP +from mcp.types import BlobResourceContents, TextResourceContents + +mcp = FastMCP(name="Direct ResourceContents Example") + + +# Example 1: Return TextResourceContents with metadata +@mcp.resource("document://report") +def get_report() -> TextResourceContents: + """Return a report with metadata about creation time and author.""" + return TextResourceContents( + uri=AnyUrl("document://report"), + text="# Monthly Report\n\nThis is the monthly report content.", + mimeType="text/markdown", + # The main benefit: adding metadata to the resource + _meta={ + "created": "2024-01-15T10:30:00Z", + "author": "Analytics Team", + "version": "1.2.0", + "tags": ["monthly", "finance", "q1-2024"], + "confidentiality": "internal", + }, + ) + + +# Example 2: Return BlobResourceContents with image metadata +@mcp.resource("image://logo") +def get_logo() -> BlobResourceContents: + """Return a logo image with metadata about dimensions and format.""" + # In a real app, you might read this from a file + image_bytes = b"\x89PNG\r\n\x1a\n..." # PNG header + + return BlobResourceContents( + uri=AnyUrl("image://logo"), + blob=base64.b64encode(image_bytes).decode(), + mimeType="image/png", + # Image-specific metadata + _meta={ + "width": 512, + "height": 512, + "format": "PNG", + "colorSpace": "sRGB", + "hasAlpha": True, + "fileSize": 24576, + "lastModified": "2024-01-10T08:00:00Z", + }, + ) + + +# Example 3: Dynamic resource with real-time metadata +@mcp.resource("data://metrics/{metric_type}") +async def get_metrics(metric_type: str) -> TextResourceContents: + """Return metrics data with metadata about collection time and source.""" + import datetime + from datetime import timezone + + # Simulate collecting metrics + metrics = {"cpu": 45.2, "memory": 78.5, "disk": 62.1} + timestamp = datetime.datetime.now(timezone.utc).isoformat() + + if metric_type == "json": + import json + + return TextResourceContents( + uri=AnyUrl(f"data://metrics/{metric_type}"), + text=json.dumps(metrics, indent=2), + mimeType="application/json", + _meta={ + "timestamp": timestamp, + "source": "system_monitor", + "interval": "5s", + "aggregation": "average", + "host": "prod-server-01", + }, + ) + elif metric_type == "csv": + csv_text = "metric,value\n" + "\n".join(f"{k},{v}" for k, v in metrics.items()) + return TextResourceContents( + uri=AnyUrl(f"data://metrics/{metric_type}"), + text=csv_text, + mimeType="text/csv", + _meta={ + "timestamp": timestamp, + "columns": ["metric", "value"], + "row_count": len(metrics), + }, + ) + else: + text = "\n".join(f"{k.upper()}: {v}%" for k, v in metrics.items()) + return TextResourceContents( + uri=AnyUrl(f"data://metrics/{metric_type}"), + text=text, + mimeType="text/plain", + _meta={ + "timestamp": timestamp, + "format": "human-readable", + }, + ) + + +# Example 4: Configuration resource with version metadata +@mcp.resource("config://app") +def get_config() -> TextResourceContents: + """Return application config with version and environment metadata.""" + import json + + config = { + "version": "1.0.0", + "features": { + "dark_mode": True, + "auto_save": False, + "language": "en", + }, + "limits": { + "max_file_size": 10485760, # 10MB + "max_connections": 100, + }, + } + + return TextResourceContents( + uri=AnyUrl("config://app"), + text=json.dumps(config, indent=2), + mimeType="application/json", + _meta={ + "version": "1.0.0", + "lastUpdated": "2024-01-15T14:30:00Z", + "environment": "production", + "schema": "https://example.com/schemas/config/v1.0", + "editable": False, + }, + ) + + +# Example 5: Database query result with execution metadata +@mcp.resource("db://query/users") +async def get_users() -> TextResourceContents: + """Return query results with execution time and row count.""" + import json + import time + + # Simulate database query + start_time = time.time() + users = [ + {"id": 1, "name": "Alice", "role": "admin"}, + {"id": 2, "name": "Bob", "role": "user"}, + {"id": 3, "name": "Charlie", "role": "user"}, + ] + execution_time = time.time() - start_time + + return TextResourceContents( + uri=AnyUrl("db://query/users"), + text=json.dumps(users, indent=2), + mimeType="application/json", + _meta={ + "query": "SELECT * FROM users", + "executionTime": f"{execution_time:.3f}s", + "rowCount": len(users), + "database": "main", + "cached": False, + "timestamp": "2024-01-15T16:00:00Z", + }, + ) + + +if __name__ == "__main__": + # Run with: python resource_contents_direct.py + mcp.run() diff --git a/src/mcp/server/fastmcp/resources/base.py b/src/mcp/server/fastmcp/resources/base.py index 0bef1a266..349580f9c 100644 --- a/src/mcp/server/fastmcp/resources/base.py +++ b/src/mcp/server/fastmcp/resources/base.py @@ -13,7 +13,7 @@ field_validator, ) -from mcp.types import Icon +from mcp.types import BlobResourceContents, Icon, TextResourceContents class Resource(BaseModel, abc.ABC): @@ -43,6 +43,6 @@ def set_default_name(cls, name: str | None, info: ValidationInfo) -> str: raise ValueError("Either name or uri must be provided") @abc.abstractmethod - async def read(self) -> str | bytes: + async def read(self) -> str | bytes | TextResourceContents | BlobResourceContents: """Read the resource content.""" pass diff --git a/src/mcp/server/fastmcp/resources/types.py b/src/mcp/server/fastmcp/resources/types.py index c578e23de..1d4b1ff53 100644 --- a/src/mcp/server/fastmcp/resources/types.py +++ b/src/mcp/server/fastmcp/resources/types.py @@ -14,7 +14,7 @@ from pydantic import AnyUrl, Field, ValidationInfo, validate_call from mcp.server.fastmcp.resources.base import Resource -from mcp.types import Icon +from mcp.types import BlobResourceContents, Icon, TextResourceContents class TextResource(Resource): @@ -52,7 +52,7 @@ class FunctionResource(Resource): fn: Callable[[], Any] = Field(exclude=True) - async def read(self) -> str | bytes: + async def read(self) -> str | bytes | TextResourceContents | BlobResourceContents: """Read the resource by calling the wrapped function.""" try: # Call the function first to see if it returns a coroutine @@ -63,6 +63,8 @@ async def read(self) -> str | bytes: if isinstance(result, Resource): return await result.read() + elif isinstance(result, TextResourceContents | BlobResourceContents): + return result elif isinstance(result, bytes): return result elif isinstance(result, str): diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 485ef1519..de916d873 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -43,7 +43,16 @@ from mcp.server.streamable_http_manager import StreamableHTTPSessionManager from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.context import LifespanContextT, RequestContext, RequestT -from mcp.types import AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations +from mcp.types import ( + AnyFunction, + BlobResourceContents, + ContentBlock, + GetPromptResult, + Icon, + ResourceContents, + TextResourceContents, + ToolAnnotations, +) from mcp.types import Prompt as MCPPrompt from mcp.types import PromptArgument as MCPPromptArgument from mcp.types import Resource as MCPResource @@ -340,7 +349,9 @@ async def list_resource_templates(self) -> list[MCPResourceTemplate]: for template in templates ] - async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]: + async def read_resource( + self, uri: AnyUrl | str + ) -> Iterable[ReadResourceContents | TextResourceContents | BlobResourceContents]: """Read a resource by URI.""" context = self.get_context() @@ -350,7 +361,10 @@ async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContent try: content = await resource.read() - return [ReadResourceContents(content=content, mime_type=resource.mime_type)] + if isinstance(content, ResourceContents): + return [content] + else: + return [ReadResourceContents(content=content, mime_type=resource.mime_type)] except Exception as e: logger.exception(f"Error reading resource {uri}") raise ResourceError(str(e)) @@ -1124,7 +1138,9 @@ async def report_progress(self, progress: float, total: float | None = None, mes message=message, ) - async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]: + async def read_resource( + self, uri: str | AnyUrl + ) -> Iterable[ReadResourceContents | TextResourceContents | BlobResourceContents]: """Read a resource by URI. Args: diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 1a69cbe48..bf4dcc186 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -311,7 +311,14 @@ async def handler(_: Any): def read_resource(self): def decorator( - func: Callable[[AnyUrl], Awaitable[str | bytes | Iterable[ReadResourceContents]]], + func: Callable[ + [AnyUrl], + Awaitable[ + str + | bytes + | Iterable[ReadResourceContents | types.TextResourceContents | types.BlobResourceContents] + ], + ], ): logger.debug("Registering handler for ReadResourceRequest") @@ -346,7 +353,10 @@ def create_content(data: str | bytes, mime_type: str | None): content = create_content(data, None) case Iterable() as contents: contents_list = [ - create_content(content_item.content, content_item.mime_type) for content_item in contents + content_item + if isinstance(content_item, types.ResourceContents) + else create_content(content_item.content, content_item.mime_type) + for content_item in contents ] return types.ServerResult( types.ReadResourceResult( diff --git a/tests/issues/test_141_resource_templates.py b/tests/issues/test_141_resource_templates.py index 3145f65e8..23c24b7c3 100644 --- a/tests/issues/test_141_resource_templates.py +++ b/tests/issues/test_141_resource_templates.py @@ -53,8 +53,21 @@ def get_user_profile_missing(user_id: str) -> str: result = await mcp.read_resource("resource://users/123/posts/456") result_list = list(result) assert len(result_list) == 1 - assert result_list[0].content == "Post 456 by user 123" - assert result_list[0].mime_type == "text/plain" + content = result_list[0] + # Since this is a string resource, it should be wrapped as ReadResourceContents + from mcp.server.lowlevel.server import ReadResourceContents + from mcp.types import TextResourceContents + + if isinstance(content, ReadResourceContents): + assert content.content == "Post 456 by user 123" + assert content.mime_type == "text/plain" + elif isinstance(content, TextResourceContents): + # If it's TextResourceContents (direct return) + assert content.text == "Post 456 by user 123" + assert content.mimeType == "text/plain" + else: + # Should not happen for string resources + raise AssertionError(f"Unexpected content type: {type(content)}") # Verify invalid parameters raise error with pytest.raises(ValueError, match="Unknown resource"): diff --git a/tests/server/fastmcp/resources/test_resource_contents_direct.py b/tests/server/fastmcp/resources/test_resource_contents_direct.py new file mode 100644 index 000000000..5fc4dee2f --- /dev/null +++ b/tests/server/fastmcp/resources/test_resource_contents_direct.py @@ -0,0 +1,188 @@ +"""Test FastMCP resources returning ResourceContents directly.""" + +import pytest +from pydantic import AnyUrl + +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.resources import Resource +from mcp.types import BlobResourceContents, TextResourceContents + + +@pytest.mark.anyio +async def test_resource_returns_text_resource_contents_directly(): + """Test a custom resource that returns TextResourceContents directly.""" + app = FastMCP("test") + + class DirectTextResource(Resource): + """A resource that returns TextResourceContents directly.""" + + async def read(self): + # Return TextResourceContents directly instead of str + return TextResourceContents( + uri=self.uri, + text="Direct TextResourceContents content", + mimeType="text/markdown", + ) + + # Add the resource + app.add_resource( + DirectTextResource( + uri=AnyUrl("resource://direct-text"), + name="direct-text", + title="Direct Text Resource", + description="Returns TextResourceContents directly", + ) + ) + + # Read the resource + contents = await app.read_resource("resource://direct-text") + contents_list = list(contents) + + # Verify the result + assert len(contents_list) == 1 + content = contents_list[0] + assert isinstance(content, TextResourceContents) + assert content.text == "Direct TextResourceContents content" + assert content.mimeType == "text/markdown" + assert str(content.uri) == "resource://direct-text" + + +@pytest.mark.anyio +async def test_resource_returns_blob_resource_contents_directly(): + """Test a custom resource that returns BlobResourceContents directly.""" + app = FastMCP("test") + + class DirectBlobResource(Resource): + """A resource that returns BlobResourceContents directly.""" + + async def read(self): + # Return BlobResourceContents directly + return BlobResourceContents( + uri=self.uri, + blob="SGVsbG8gRmFzdE1DUA==", # "Hello FastMCP" in base64 + mimeType="application/pdf", + ) + + # Add the resource + app.add_resource( + DirectBlobResource( + uri=AnyUrl("resource://direct-blob"), + name="direct-blob", + title="Direct Blob Resource", + description="Returns BlobResourceContents directly", + ) + ) + + # Read the resource + contents = await app.read_resource("resource://direct-blob") + contents_list = list(contents) + + # Verify the result + assert len(contents_list) == 1 + content = contents_list[0] + assert isinstance(content, BlobResourceContents) + assert content.blob == "SGVsbG8gRmFzdE1DUA==" + assert content.mimeType == "application/pdf" + assert str(content.uri) == "resource://direct-blob" + + +@pytest.mark.anyio +async def test_function_resource_returns_resource_contents(): + """Test function resource returning ResourceContents directly.""" + app = FastMCP("test") + + @app.resource("resource://function-text-contents") + async def get_text_contents() -> TextResourceContents: + """Return TextResourceContents directly from function resource.""" + return TextResourceContents( + uri=AnyUrl("resource://function-text-contents"), + text="Function returned TextResourceContents", + mimeType="text/x-python", + ) + + @app.resource("resource://function-blob-contents") + def get_blob_contents() -> BlobResourceContents: + """Return BlobResourceContents directly from function resource.""" + return BlobResourceContents( + uri=AnyUrl("resource://function-blob-contents"), + blob="RnVuY3Rpb24gYmxvYg==", # "Function blob" in base64 + mimeType="image/png", + ) + + # Read text resource + text_contents = await app.read_resource("resource://function-text-contents") + text_list = list(text_contents) + assert len(text_list) == 1 + text_content = text_list[0] + assert isinstance(text_content, TextResourceContents) + assert text_content.text == "Function returned TextResourceContents" + assert text_content.mimeType == "text/x-python" + + # Read blob resource + blob_contents = await app.read_resource("resource://function-blob-contents") + blob_list = list(blob_contents) + assert len(blob_list) == 1 + blob_content = blob_list[0] + assert isinstance(blob_content, BlobResourceContents) + assert blob_content.blob == "RnVuY3Rpb24gYmxvYg==" + assert blob_content.mimeType == "image/png" + + +@pytest.mark.anyio +async def test_mixed_traditional_and_direct_resources(): + """Test server with both traditional and direct ResourceContents resources.""" + app = FastMCP("test") + + # Traditional string resource + @app.resource("resource://traditional") + def traditional_resource() -> str: + return "Traditional string content" + + # Direct ResourceContents resource + @app.resource("resource://direct") + def direct_resource() -> TextResourceContents: + return TextResourceContents( + uri=AnyUrl("resource://direct"), + text="Direct ResourceContents content", + mimeType="text/html", + ) + + # Read traditional resource (will be wrapped) + trad_contents = await app.read_resource("resource://traditional") + trad_list = list(trad_contents) + assert len(trad_list) == 1 + # The content type might be ReadResourceContents, but we're checking the behavior + + # Read direct ResourceContents + direct_contents = await app.read_resource("resource://direct") + direct_list = list(direct_contents) + assert len(direct_list) == 1 + direct_content = direct_list[0] + assert isinstance(direct_content, TextResourceContents) + assert direct_content.text == "Direct ResourceContents content" + assert direct_content.mimeType == "text/html" + + +@pytest.mark.anyio +async def test_resource_template_returns_resource_contents(): + """Test resource template returning ResourceContents directly.""" + app = FastMCP("test") + + @app.resource("resource://{category}/{item}") + async def get_item_contents(category: str, item: str) -> TextResourceContents: + """Return TextResourceContents for template resource.""" + return TextResourceContents( + uri=AnyUrl(f"resource://{category}/{item}"), + text=f"Content for {item} in {category}", + mimeType="text/plain", + ) + + # Read templated resource + contents = await app.read_resource("resource://books/python") + contents_list = list(contents) + assert len(contents_list) == 1 + content = contents_list[0] + assert isinstance(content, TextResourceContents) + assert content.text == "Content for python in books" + assert content.mimeType == "text/plain" + assert str(content.uri) == "resource://books/python" diff --git a/tests/server/fastmcp/servers/test_file_server.py b/tests/server/fastmcp/servers/test_file_server.py index df7024552..ca9653060 100644 --- a/tests/server/fastmcp/servers/test_file_server.py +++ b/tests/server/fastmcp/servers/test_file_server.py @@ -92,9 +92,19 @@ async def test_read_resource_dir(mcp: FastMCP): res_list = list(res_iter) assert len(res_list) == 1 res = res_list[0] - assert res.mime_type == "text/plain" - files = json.loads(res.content) + # Handle union type properly + from mcp.server.lowlevel.server import ReadResourceContents + from mcp.types import TextResourceContents + + if isinstance(res, ReadResourceContents): + assert res.mime_type == "text/plain" + files = json.loads(res.content) + elif isinstance(res, TextResourceContents): + assert res.mimeType == "text/plain" + files = json.loads(res.text) + else: + raise AssertionError(f"Unexpected content type: {type(res)}") assert sorted([Path(f).name for f in files]) == [ "config.json", @@ -109,7 +119,17 @@ async def test_read_resource_file(mcp: FastMCP): res_list = list(res_iter) assert len(res_list) == 1 res = res_list[0] - assert res.content == "print('hello world')" + + # Handle union type properly + from mcp.server.lowlevel.server import ReadResourceContents + from mcp.types import TextResourceContents + + if isinstance(res, ReadResourceContents): + assert res.content == "print('hello world')" + elif isinstance(res, TextResourceContents): + assert res.text == "print('hello world')" + else: + raise AssertionError(f"Unexpected content type: {type(res)}") @pytest.mark.anyio @@ -125,4 +145,14 @@ async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path): res_list = list(res_iter) assert len(res_list) == 1 res = res_list[0] - assert res.content == "File not found" + + # Handle union type properly + from mcp.server.lowlevel.server import ReadResourceContents + from mcp.types import TextResourceContents + + if isinstance(res, ReadResourceContents): + assert res.content == "File not found" + elif isinstance(res, TextResourceContents): + assert res.text == "File not found" + else: + raise AssertionError(f"Unexpected content type: {type(res)}") diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 8caa3b1f6..bd26e16bb 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -1035,7 +1035,16 @@ async def tool_with_resource(ctx: Context[ServerSession, None]) -> str: r_list = list(r_iter) assert len(r_list) == 1 r = r_list[0] - return f"Read resource: {r.content} with mime type {r.mime_type}" + # Handle union type properly + from mcp.server.lowlevel.server import ReadResourceContents + from mcp.types import TextResourceContents + + if isinstance(r, ReadResourceContents): + return f"Read resource: {r.content} with mime type {r.mime_type}" + elif isinstance(r, TextResourceContents): + return f"Read resource: {r.text} with mime type {r.mimeType}" + else: + raise AssertionError(f"Unexpected content type: {type(r)}") async with client_session(mcp._mcp_server) as client: result = await client.call_tool("tool_with_resource", {}) diff --git a/tests/server/test_read_resource_direct.py b/tests/server/test_read_resource_direct.py new file mode 100644 index 000000000..2a7643b61 --- /dev/null +++ b/tests/server/test_read_resource_direct.py @@ -0,0 +1,191 @@ +from collections.abc import Iterable +from pathlib import Path +from tempfile import NamedTemporaryFile + +import pytest +from pydantic import AnyUrl, FileUrl + +import mcp.types as types +from mcp.server.lowlevel.server import ReadResourceContents, Server + + +@pytest.fixture +def temp_file(): + """Create a temporary file for testing.""" + with NamedTemporaryFile(mode="w", delete=False) as f: + f.write("test content") + path = Path(f.name).resolve() + yield path + try: + path.unlink() + except FileNotFoundError: + pass + + +@pytest.mark.anyio +async def test_read_resource_direct_text_resource_contents(temp_file: Path): + """Test returning TextResourceContents directly from read_resource handler.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents]: + return [ + types.TextResourceContents( + uri=uri, + text="Direct text content", + mimeType="text/markdown", + ) + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.TextResourceContents) + assert content.text == "Direct text content" + assert content.mimeType == "text/markdown" + assert str(content.uri) == temp_file.as_uri() + + +@pytest.mark.anyio +async def test_read_resource_direct_blob_resource_contents(temp_file: Path): + """Test returning BlobResourceContents directly from read_resource handler.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[types.BlobResourceContents]: + return [ + types.BlobResourceContents( + uri=uri, + blob="SGVsbG8gV29ybGQ=", # "Hello World" in base64 + mimeType="application/pdf", + ) + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.BlobResourceContents) + assert content.blob == "SGVsbG8gV29ybGQ=" + assert content.mimeType == "application/pdf" + assert str(content.uri) == temp_file.as_uri() + + +@pytest.mark.anyio +async def test_read_resource_mixed_contents(temp_file: Path): + """Test mixing direct ResourceContents with ReadResourceContents.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents | types.TextResourceContents]: + return [ + types.TextResourceContents( + uri=uri, + text="Direct ResourceContents", + mimeType="text/plain", + ), + ReadResourceContents(content="Wrapped content", mime_type="text/html"), + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 2 + + # First content is direct ResourceContents + content1 = result.root.contents[0] + assert isinstance(content1, types.TextResourceContents) + assert content1.text == "Direct ResourceContents" + assert content1.mimeType == "text/plain" + + # Second content is wrapped ReadResourceContents + content2 = result.root.contents[1] + assert isinstance(content2, types.TextResourceContents) + assert content2.text == "Wrapped content" + assert content2.mimeType == "text/html" + + +@pytest.mark.anyio +async def test_read_resource_multiple_resource_contents(temp_file: Path): + """Test returning multiple ResourceContents objects.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | types.BlobResourceContents]: + return [ + types.TextResourceContents( + uri=uri, + text="First text content", + mimeType="text/plain", + ), + types.BlobResourceContents( + uri=uri, + blob="U2Vjb25kIGNvbnRlbnQ=", # "Second content" in base64 + mimeType="application/octet-stream", + ), + types.TextResourceContents( + uri=uri, + text="Third text content", + mimeType="text/markdown", + ), + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 3 + + # Check first content + content1 = result.root.contents[0] + assert isinstance(content1, types.TextResourceContents) + assert content1.text == "First text content" + assert content1.mimeType == "text/plain" + + # Check second content + content2 = result.root.contents[1] + assert isinstance(content2, types.BlobResourceContents) + assert content2.blob == "U2Vjb25kIGNvbnRlbnQ=" + assert content2.mimeType == "application/octet-stream" + + # Check third content + content3 = result.root.contents[2] + assert isinstance(content3, types.TextResourceContents) + assert content3.text == "Third text content" + assert content3.mimeType == "text/markdown"