Allow MCPServer.read_resource() to read resources by Resource.

Author: fennb

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -10,7 +10,7 @@
 from dataclasses import field, replace
 from datetime import timedelta
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated, Any, overload
 
 import anyio
 import httpx
@@ -320,18 +320,29 @@ async def list_resource_templates(self) -> list[_mcp.ResourceTemplate]:
             result = await self._client.list_resource_templates()
         return [_mcp.map_from_mcp_resource_template(t) for t in result.resourceTemplates]
 
-    async def read_resource(self, uri: str) -> str | messages.BinaryContent | list[str | messages.BinaryContent]:
+    @overload
+    async def read_resource(self, uri: str) -> str | messages.BinaryContent | list[str | messages.BinaryContent]: ...
+
+    @overload
+    async def read_resource(
+        self, uri: _mcp.Resource
+    ) -> str | messages.BinaryContent | list[str | messages.BinaryContent]: ...
+
+    async def read_resource(
+        self, uri: str | _mcp.Resource
+    ) -> str | messages.BinaryContent | list[str | messages.BinaryContent]:
         """Read the contents of a specific resource by URI.
 
         Args:
-            uri: The URI of the resource to read.
+            uri: The URI of the resource to read, or a Resource object.
 
         Returns:
             The resource contents. If the resource has a single content item, returns that item directly.
             If the resource has multiple content items, returns a list of items.
         """
+        resource_uri = uri if isinstance(uri, str) else uri.uri
         async with self:  # Ensure server is running
-            result = await self._client.read_resource(AnyUrl(uri))
+            result = await self._client.read_resource(AnyUrl(resource_uri))
         return (
             self._get_content(result.contents[0])
             if len(result.contents) == 1

tests/test_mcp.py

@@ -23,6 +23,7 @@
     ToolReturnPart,
     UserPromptPart,
 )
+from pydantic_ai._mcp import Resource
 from pydantic_ai.agent import Agent
 from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior, UserError
 from pydantic_ai.mcp import MCPServerStreamableHTTP, load_mcp_servers
@@ -1500,10 +1501,17 @@ async def test_read_text_resource(run_context: RunContext[int]):
     """Test reading a text resource (converted to string)."""
     server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
     async with server:
+        # Test reading by URI string
         content = await server.read_resource('resource://product_name.txt')
         assert isinstance(content, str)
         assert content == snapshot('Pydantic AI\n')
 
+        # Test reading by Resource object
+        resource = Resource(uri='resource://product_name.txt', name='product_name_resource')
+        content_from_resource = await server.read_resource(resource)
+        assert isinstance(content_from_resource, str)
+        assert content_from_resource == snapshot('Pydantic AI\n')
+
 
 async def test_read_blob_resource(run_context: RunContext[int]):
     """Test reading a binary resource (converted to BinaryContent)."""