feat(mcp): expose list_resources, list_resource_templates, and read_resource on MCPServer (#2721)

Author: adityasingh2400

src/agents/mcp/server.py

@@ -22,7 +22,15 @@
 from mcp.client.streamable_http import GetSessionIdCallback, streamablehttp_client
 from mcp.shared.exceptions import McpError
 from mcp.shared.message import SessionMessage
-from mcp.types import CallToolResult, GetPromptResult, InitializeResult, ListPromptsResult
+from mcp.types import (
+    CallToolResult,
+    GetPromptResult,
+    InitializeResult,
+    ListPromptsResult,
+    ListResourcesResult,
+    ListResourceTemplatesResult,
+    ReadResourceResult,
+)
 from typing_extensions import NotRequired, TypedDict
 
 from ..exceptions import UserError
@@ -192,6 +200,63 @@ async def get_prompt(
         """Get a specific prompt from the server."""
         pass
 
+    async def list_resources(self, cursor: str | None = None) -> ListResourcesResult:
+        """List the resources available on the server.
+
+        Args:
+            cursor: An opaque pagination cursor returned in a previous
+                :class:`~mcp.types.ListResourcesResult` as ``nextCursor``.  Pass it
+                here to fetch the next page of results.  ``None`` fetches the first
+                page.
+
+        Returns a :class:`~mcp.types.ListResourcesResult`.  When the result contains
+        a ``nextCursor`` field, call this method again with that cursor to retrieve
+        the next page.  Subclasses that do not support resources may leave this
+        unimplemented; it will raise :exc:`NotImplementedError` at call time.
+        """
+        raise NotImplementedError(
+            f"MCP server '{self.name}' does not support list_resources. "
+            "Override this method in your server implementation."
+        )
+
+    async def list_resource_templates(
+        self, cursor: str | None = None
+    ) -> ListResourceTemplatesResult:
+        """List the resource templates available on the server.
+
+        Args:
+            cursor: An opaque pagination cursor returned in a previous
+                :class:`~mcp.types.ListResourceTemplatesResult` as ``nextCursor``.
+                Pass it here to fetch the next page of results.  ``None`` fetches
+                the first page.
+
+        Returns a :class:`~mcp.types.ListResourceTemplatesResult`.  When the result
+        contains a ``nextCursor`` field, call this method again with that cursor to
+        retrieve the next page.  Subclasses that do not support resource templates
+        may leave this unimplemented; it will raise :exc:`NotImplementedError` at
+        call time.
+        """
+        raise NotImplementedError(
+            f"MCP server '{self.name}' does not support list_resource_templates. "
+            "Override this method in your server implementation."
+        )
+
+    async def read_resource(self, uri: str) -> ReadResourceResult:
+        """Read the contents of a specific resource by URI.
+
+        Args:
+            uri: The URI of the resource to read. See :class:`~pydantic.networks.AnyUrl`
+                for the supported URI formats.
+
+        Returns a :class:`~mcp.types.ReadResourceResult`.  Subclasses that do not
+        support resources may leave this unimplemented; it will raise
+        :exc:`NotImplementedError` at call time.
+        """
+        raise NotImplementedError(
+            f"MCP server '{self.name}' does not support read_resource. "
+            "Override this method in your server implementation."
+        )
+
     @staticmethod
     def _normalize_needs_approval(
         *,
@@ -708,6 +773,39 @@ async def get_prompt(
         assert session is not None
         return await self._maybe_serialize_request(lambda: session.get_prompt(name, arguments))
 
+    async def list_resources(self, cursor: str | None = None) -> ListResourcesResult:
+        """List the resources available on the server."""
+        if not self.session:
+            raise UserError("Server not initialized. Make sure you call `connect()` first.")
+        session = self.session
+        assert session is not None
+        return await self._maybe_serialize_request(lambda: session.list_resources(cursor))
+
+    async def list_resource_templates(
+        self, cursor: str | None = None
+    ) -> ListResourceTemplatesResult:
+        """List the resource templates available on the server."""
+        if not self.session:
+            raise UserError("Server not initialized. Make sure you call `connect()` first.")
+        session = self.session
+        assert session is not None
+        return await self._maybe_serialize_request(lambda: session.list_resource_templates(cursor))
+
+    async def read_resource(self, uri: str) -> ReadResourceResult:
+        """Read the contents of a specific resource by URI.
+
+        Args:
+            uri: The URI of the resource to read. See :class:`~pydantic.networks.AnyUrl`
+                for the supported URI formats.
+        """
+        if not self.session:
+            raise UserError("Server not initialized. Make sure you call `connect()` first.")
+        session = self.session
+        assert session is not None
+        from pydantic import AnyUrl
+
+        return await self._maybe_serialize_request(lambda: session.read_resource(AnyUrl(uri)))
+
     async def cleanup(self):
         """Cleanup the server."""
         async with self._cleanup_lock:

tests/mcp/helpers.py

@@ -11,7 +11,10 @@
     Content,
     GetPromptResult,
     ListPromptsResult,
+    ListResourcesResult,
+    ListResourceTemplatesResult,
     PromptMessage,
+    ReadResourceResult,
     TextContent,
 )
 
@@ -138,6 +141,20 @@ async def get_prompt(
         message = PromptMessage(role="user", content=TextContent(type="text", text=content))
         return GetPromptResult(description=f"Fake prompt: {name}", messages=[message])
 
+    async def list_resources(self, cursor: str | None = None) -> ListResourcesResult:
+        """Return empty list of resources for fake server."""
+        return ListResourcesResult(resources=[])
+
+    async def list_resource_templates(
+        self, cursor: str | None = None
+    ) -> ListResourceTemplatesResult:
+        """Return empty list of resource templates for fake server."""
+        return ListResourceTemplatesResult(resourceTemplates=[])
+
+    async def read_resource(self, uri: str) -> ReadResourceResult:
+        """Return empty resource contents for fake server."""
+        return ReadResourceResult(contents=[])
+
     @property
     def name(self) -> str:
         return self._server_name

tests/mcp/test_mcp_resources.py

@@ -0,0 +1,175 @@
+"""Tests for MCP server list_resources, list_resource_templates, and read_resource."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from mcp.types import (
+    ListResourcesResult,
+    ListResourceTemplatesResult,
+    ReadResourceResult,
+    Resource,
+    ResourceTemplate,
+    TextResourceContents,
+)
+from pydantic import AnyUrl
+
+from agents.mcp import MCPServerStreamableHttp
+
+
+@pytest.fixture
+def server():
+    return MCPServerStreamableHttp(params={"url": "http://localhost:8000/mcp"})
+
+
+@pytest.mark.asyncio
+async def test_list_resources_raises_when_not_connected(server: MCPServerStreamableHttp):
+    """list_resources raises UserError when server has not been connected."""
+    from agents.exceptions import UserError
+
+    with pytest.raises(UserError, match="Server not initialized"):
+        await server.list_resources()
+
+
+@pytest.mark.asyncio
+async def test_list_resource_templates_raises_when_not_connected(server: MCPServerStreamableHttp):
+    """list_resource_templates raises UserError when server has not been connected."""
+    from agents.exceptions import UserError
+
+    with pytest.raises(UserError, match="Server not initialized"):
+        await server.list_resource_templates()
+
+
+@pytest.mark.asyncio
+async def test_read_resource_raises_when_not_connected(server: MCPServerStreamableHttp):
+    """read_resource raises UserError when server has not been connected."""
+    from agents.exceptions import UserError
+
+    with pytest.raises(UserError, match="Server not initialized"):
+        await server.read_resource("file:///etc/hosts")
+
+
+@pytest.mark.asyncio
+async def test_list_resources_returns_result(server: MCPServerStreamableHttp):
+    """list_resources delegates to the underlying MCP session."""
+    mock_session = MagicMock()
+    expected = ListResourcesResult(
+        resources=[
+            Resource(uri=AnyUrl("file:///readme.md"), name="readme.md", mimeType="text/markdown"),
+        ]
+    )
+    mock_session.list_resources = AsyncMock(return_value=expected)
+    server.session = mock_session
+
+    result = await server.list_resources()
+
+    assert result is expected
+    mock_session.list_resources.assert_awaited_once_with(None)
+
+
+@pytest.mark.asyncio
+async def test_list_resources_forwards_cursor(server: MCPServerStreamableHttp):
+    """list_resources forwards the cursor argument for pagination."""
+    mock_session = MagicMock()
+    page2 = ListResourcesResult(resources=[])
+    mock_session.list_resources = AsyncMock(return_value=page2)
+    server.session = mock_session
+
+    result = await server.list_resources(cursor="tok_abc")
+
+    assert result is page2
+    mock_session.list_resources.assert_awaited_once_with("tok_abc")
+
+
+@pytest.mark.asyncio
+async def test_list_resource_templates_returns_result(server: MCPServerStreamableHttp):
+    """list_resource_templates delegates to the underlying MCP session."""
+    mock_session = MagicMock()
+    expected = ListResourceTemplatesResult(
+        resourceTemplates=[
+            ResourceTemplate(uriTemplate="file:///{path}", name="file"),
+        ]
+    )
+    mock_session.list_resource_templates = AsyncMock(return_value=expected)
+    server.session = mock_session
+
+    result = await server.list_resource_templates()
+
+    assert result is expected
+    mock_session.list_resource_templates.assert_awaited_once_with(None)
+
+
+@pytest.mark.asyncio
+async def test_list_resource_templates_forwards_cursor(server: MCPServerStreamableHttp):
+    """list_resource_templates forwards the cursor argument for pagination."""
+    mock_session = MagicMock()
+    page2 = ListResourceTemplatesResult(resourceTemplates=[])
+    mock_session.list_resource_templates = AsyncMock(return_value=page2)
+    server.session = mock_session
+
+    result = await server.list_resource_templates(cursor="tok_xyz")
+
+    assert result is page2
+    mock_session.list_resource_templates.assert_awaited_once_with("tok_xyz")
+
+
+@pytest.mark.asyncio
+async def test_read_resource_returns_result(server: MCPServerStreamableHttp):
+    """read_resource delegates to the underlying MCP session with the given URI."""
+    mock_session = MagicMock()
+    uri = "file:///readme.md"
+    expected = ReadResourceResult(
+        contents=[
+            TextResourceContents(uri=AnyUrl(uri), text="# Hello", mimeType="text/markdown"),
+        ]
+    )
+    mock_session.read_resource = AsyncMock(return_value=expected)
+    server.session = mock_session
+
+    result = await server.read_resource(uri)
+
+    assert result is expected
+    mock_session.read_resource.assert_awaited_once_with(AnyUrl(uri))
+
+
+@pytest.mark.asyncio
+async def test_base_methods_raise_not_implemented():
+    """Bare MCPServer subclasses that don't override resource methods get NotImplementedError."""
+    from mcp.types import CallToolResult, GetPromptResult, ListPromptsResult
+
+    from agents.mcp import MCPServer
+
+    class MinimalServer(MCPServer):
+        """Minimal subclass implementing only the truly abstract methods."""
+
+        @property
+        def name(self) -> str:
+            return "minimal"
+
+        async def connect(self) -> None:
+            pass
+
+        async def cleanup(self) -> None:
+            pass
+
+        async def list_tools(self, run_context=None, agent=None):
+            return []
+
+        async def call_tool(self, tool_name, tool_arguments, run_context=None, agent=None):
+            return CallToolResult(content=[])
+
+        async def list_prompts(self):
+            return ListPromptsResult(prompts=[])
+
+        async def get_prompt(self, name, arguments=None):
+            return GetPromptResult(messages=[])
+
+    s = MinimalServer()
+
+    with pytest.raises(NotImplementedError, match="list_resources"):
+        await s.list_resources()
+
+    with pytest.raises(NotImplementedError, match="list_resource_templates"):
+        await s.list_resource_templates()
+
+    with pytest.raises(NotImplementedError, match="read_resource"):
+        await s.read_resource("file:///test.txt")