Add resource Link (#974)

examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py

@@ -41,7 +41,7 @@ def main(
     app = Server("mcp-streamable-http-stateless-demo")
 
     @app.call_tool()
-    async def call_tool(name: str, arguments: dict) -> list[types.Content]:
+    async def call_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
         ctx = app.request_context
         interval = arguments.get("interval", 1.0)
         count = arguments.get("count", 5)

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

@@ -45,7 +45,7 @@ def main(
     app = Server("mcp-streamable-http-demo")
 
     @app.call_tool()
-    async def call_tool(name: str, arguments: dict) -> list[types.Content]:
+    async def call_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
         ctx = app.request_context
         interval = arguments.get("interval", 1.0)
         count = arguments.get("count", 5)

examples/servers/simple-tool/mcp_simple_tool/server.py

@@ -7,7 +7,7 @@ from mcp.shared._httpx_utils import create_mcp_http_client
 
 async def fetch_website(
     url: str,
-) -> list[types.Content]:
+) -> list[types.ContentBlock]:
     headers = {
         "User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
     }
@@ -29,7 +29,7 @@ def main(port: int, transport: str) -> int:
     app = Server("mcp-website-fetcher")
 
     @app.call_tool()
-    async def fetch_tool(name: str, arguments: dict) -> list[types.Content]:
+    async def fetch_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
         if name != "fetch":
             raise ValueError(f"Unknown tool: {name}")
         if "url" not in arguments:

src/mcp/server/fastmcp/prompts/base.py

@@ -7,16 +7,16 @@ from typing import Any, Literal
 import pydantic_core
 from pydantic import BaseModel, Field, TypeAdapter, validate_call
 
-from mcp.types import Content, TextContent
+from mcp.types import ContentBlock, TextContent
 
 
 class Message(BaseModel):
     """Base class for all prompt messages."""
 
     role: Literal["user", "assistant"]
-    content: Content
+    content: ContentBlock
 
-    def __init__(self, content: str | Content, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock, **kwargs: Any):
         if isinstance(content, str):
             content = TextContent(type="text", text=content)
         super().__init__(content=content, **kwargs)
@@ -27,7 +27,7 @@ class UserMessage(Message):
 
     role: Literal["user", "assistant"] = "user"
 
-    def __init__(self, content: str | Content, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock, **kwargs: Any):
         super().__init__(content=content, **kwargs)
 
 
@@ -36,7 +36,7 @@ class AssistantMessage(Message):
 
     role: Literal["user", "assistant"] = "assistant"
 
-    def __init__(self, content: str | Content, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock, **kwargs: Any):
         super().__init__(content=content, **kwargs)
 
 

src/mcp/server/fastmcp/server.py

@@ -54,7 +54,7 @@ from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import (
     AnyFunction,
-    Content,
+    ContentBlock,
     GetPromptResult,
     TextContent,
     ToolAnnotations,
@@ -260,7 +260,7 @@ class FastMCP:
             request_context = None
         return Context(request_context=request_context, fastmcp=self)
 
-    async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[Content]:
+    async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock]:
         """Call a tool by name with arguments."""
         context = self.get_context()
         result = await self._tool_manager.call_tool(name, arguments, context=context)
@@ -878,12 +878,12 @@ class FastMCP:
 
 def _convert_to_content(
     result: Any,
-) -> Sequence[Content]:
+) -> Sequence[ContentBlock]:
     """Convert a result to a sequence of content objects."""
     if result is None:
         return []
 
-    if isinstance(result, Content):
+    if isinstance(result, ContentBlock):
         return [result]
 
     if isinstance(result, Image):

src/mcp/server/lowlevel/server.py

@@ -384,7 +384,7 @@ class Server(Generic[LifespanResultT, RequestT]):
         def decorator(
             func: Callable[
                 ...,
-                Awaitable[Iterable[types.Content]],
+                Awaitable[Iterable[types.ContentBlock]],
             ],
         ):
             logger.debug("Registering handler for CallToolRequest")

src/mcp/types.py

@@ -733,14 +733,28 @@ class EmbeddedResource(BaseModel):
     model_config = ConfigDict(extra="allow")
 
 
-Content = TextContent | ImageContent | AudioContent | EmbeddedResource
+class ResourceLink(Resource):
+    """
+    A resource that the server is capable of reading, included in a prompt or tool call result.
+
+    Note: resource links returned by tools are not guaranteed to appear in the results of `resources/list` requests.
+    """
+
+    type: Literal["resource_link"]
+
+
+ContentBlock = TextContent | ImageContent | AudioContent | ResourceLink | EmbeddedResource
+"""A content block that can be used in prompts and tool results."""
+
+Content: TypeAlias = ContentBlock
+# """DEPRECATED: Content is deprecated, you should use ContentBlock directly."""
 
 
 class PromptMessage(BaseModel):
     """Describes a message returned as part of a prompt."""
 
     role: Role
-    content: Content
+    content: ContentBlock
     model_config = ConfigDict(extra="allow")
 
 
@@ -859,7 +873,7 @@ class CallToolRequest(Request[CallToolRequestParams, Literal["tools/call"]]):
 class CallToolResult(Result):
     """The server's response to a tool call."""
 
-    content: list[Content]
+    content: list[ContentBlock]
     isError: bool = False
 
 

tests/issues/test_88_random_error.py

@@ -11,7 +11,7 @@ from anyio.abc import TaskStatus
 from mcp.client.session import ClientSession
 from mcp.server.lowlevel import Server
 from mcp.shared.exceptions import McpError
-from mcp.types import Content, TextContent
+from mcp.types import ContentBlock, TextContent
 
 
 @pytest.mark.anyio
@@ -31,7 +31,7 @@ async def test_notification_validation_error(tmp_path: Path):
     slow_request_complete = anyio.Event()
 
     @server.call_tool()
-    async def slow_tool(name: str, arg) -> Sequence[Content]:
+    async def slow_tool(name: str, arg) -> Sequence[ContentBlock]:
         nonlocal request_count
         request_count += 1
 

tests/server/fastmcp/test_integration.py

@@ -38,6 +38,7 @@ from mcp.types import (
     ProgressNotification,
     PromptReference,
     ReadResourceResult,
+    ResourceLink,
     ResourceListChangedNotification,
     ResourceTemplateReference,
     SamplingMessage,
@@ -152,6 +153,25 @@ def make_everything_fastmcp() -> FastMCP:
     def echo(message: str) -> str:
         return f"Echo: {message}"
 
+    # Tool that returns ResourceLinks
+    @mcp.tool(description="Lists files and returns resource links", title="List Files Tool")
+    def list_files() -> list[ResourceLink]:
+        """Returns a list of resource links for files matching the pattern."""
+
+        # Mock some file resources for testing
+        file_resources = [
+            {
+                "type": "resource_link",
+                "uri": "file:///project/README.md",
+                "name": "README.md",
+                "mimeType": "text/markdown",
+            }
+        ]
+
+        result: list[ResourceLink] = [ResourceLink.model_validate(file_json) for file_json in file_resources]
+
+        return result
+
     # Tool with sampling capability
     @mcp.tool(description="A tool that uses sampling to generate content", title="Sampling Tool")
     async def sampling_tool(prompt: str, ctx: Context) -> str:
@@ -762,7 +782,17 @@ async def call_all_mcp_features(session: ClientSession, collector: NotificationC
     assert isinstance(tool_result.content[0], TextContent)
     assert tool_result.content[0].text == "Echo: hello"
 
-    # 2. Tool with context (logging and progress)
+    # 2. Test tool that returns ResourceLinks
+    list_files_result = await session.call_tool("list_files")
+    assert len(list_files_result.content) == 1
+
+    # Rest should be ResourceLinks
+    content = list_files_result.content[0]
+    assert isinstance(content, ResourceLink)
+    assert str(content.uri).startswith("file:///")
+    assert content.name is not None
+    assert content.mimeType is not None
+
     # Test progress callback functionality
     progress_updates = []
 

tests/server/fastmcp/test_server.py

@@ -18,7 +18,7 @@ from mcp.shared.memory import (
 from mcp.types import (
     AudioContent,
     BlobResourceContents,
-    Content,
+    ContentBlock,
     EmbeddedResource,
     ImageContent,
     TextContent,
@@ -194,7 +194,7 @@ def image_tool_fn(path: str) -> Image:
     return Image(path)
 
 
-def mixed_content_tool_fn() -> list[Content]:
+def mixed_content_tool_fn() -> list[ContentBlock]:
     return [
         TextContent(type="text", text="Hello"),
         ImageContent(type="image", data="abc", mimeType="image/png"),