Feature: MCP client Resources support

docs/mcp/client.md

@@ -322,6 +322,90 @@ agent = Agent('openai:gpt-4o', toolsets=[weather_server, calculator_server])
 
 MCP tools can include metadata that provides additional information about the tool's characteristics, which can be useful when [filtering tools][pydantic_ai.toolsets.FilteredToolset]. The `meta`, `annotations`, and `output_schema` fields can be found on the `metadata` dict on the [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] object that's passed to filter functions.
 
+## Resources
+
+MCP servers can provide [resources](https://modelcontextprotocol.io/docs/concepts/resources) - files, data, or content that can be accessed by the client. Resources in MCP are designed to be application-driven, with host applications determining how to incorporate context based on their needs.
+
+Pydantic AI provides methods to discover and read resources from MCP servers:
+
+- [`list_resources()`][pydantic_ai.mcp.MCPServer.list_resources] - List all available resources on the server
+- [`list_resource_templates()`][pydantic_ai.mcp.MCPServer.list_resource_templates] - List resource templates with parameter placeholders
+- [`read_resource(uri)`][pydantic_ai.mcp.MCPServer.read_resource] - Read the contents of a specific resource by URI
+
+Resources can contain either text content ([`TextResourceContents`][mcp.types.TextResourceContents]) or binary content ([`BlobResourceContents`][mcp.types.BlobResourceContents]) encoded as base64.
+
+Before consuming resources, we need to run a server that exposes some:
+
+```python {title="mcp_resource_server.py"}
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP('Pydantic AI MCP Server')
+log_level = 'unset'
+
+
+@mcp.resource('resource://user_name.txt', mime_type='text/plain')
+async def user_name_resource() -> str:
+    return 'Alice'
+
+
+if __name__ == '__main__':
+    mcp.run()
+```
+
+Then we can create the client:
+
+```python {title="mcp_resources.py", requires="mcp_resource_server.py"}
+import asyncio
+
+from mcp.types import TextResourceContents
+
+from pydantic_ai import Agent
+from pydantic_ai._run_context import RunContext
+from pydantic_ai.mcp import MCPServerStdio
+from pydantic_ai.models.test import TestModel
+
+agent = Agent(
+    model=TestModel(),
+    deps_type=str,
+    instructions="Use the customer's name while replying to them.",
+)
+
+
+@agent.instructions
+def add_the_users_name(ctx: RunContext[str]) -> str:
+    return f"The user's name is {ctx.deps}."
+
+
+async def main():
+    server = MCPServerStdio('python', args=['-m', 'mcp_resource_server'])
+
+    async with server:
+        # List all available resources
+        resources = await server.list_resources()
+        for resource in resources:
+            print(f' - {resource.name}: {resource.uri} ({resource.mimeType})')
+            #>  - user_name_resource: resource://user_name.txt (text/plain)
+
+        # Read a text resource
+        text_contents = await server.read_resource('resource://user_name.txt')
+        for content in text_contents:
+            if isinstance(content, TextResourceContents):
+                print(f'Text content from {content.uri}: {content.text.strip()}')
+                #> Text content from resource://user_name.txt: Alice
+
+    # Use resources in dependencies
+    async with agent:
+        user_name = text_contents[0].text
+        _ = await agent.run('Can you help me with my product?', deps=user_name)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())
+```
+
+_(This example is complete, it can be run "as is")_
+
+
 ## Custom TLS / SSL configuration
 
 In some environments you need to tweak how HTTPS connections are established –

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -16,7 +16,7 @@
 import httpx
 import pydantic_core
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
-from pydantic import BaseModel, Discriminator, Field, Tag
+from pydantic import AnyUrl, BaseModel, Discriminator, Field, Tag
 from pydantic_core import CoreSchema, core_schema
 from typing_extensions import Self, assert_never, deprecated
 
@@ -275,6 +275,36 @@ def tool_for_tool_def(self, tool_def: ToolDefinition) -> ToolsetTool[Any]:
             args_validator=TOOL_SCHEMA_VALIDATOR,
         )
 
+    async def list_resources(self) -> list[mcp_types.Resource]:
+        """Retrieve resources that are currently present on the server.
+
+        Note:
+        - We don't cache resources as they might change.
+        - We also don't subscribe to resource changes to avoid complexity.
+        """
+        async with self:  # Ensure server is running
+            result = await self._client.list_resources()
+        return result.resources
+
+    async def list_resource_templates(self) -> list[mcp_types.ResourceTemplate]:
+        """Retrieve resource templates that are currently present on the server."""
+        async with self:  # Ensure server is running
+            result = await self._client.list_resource_templates()
+        return result.resourceTemplates
+
+    async def read_resource(self, uri: str) -> list[mcp_types.TextResourceContents | mcp_types.BlobResourceContents]:
+        """Read the contents of a specific resource by URI.
+
+        Args:
+            uri: The URI of the resource to read.
+
+        Returns:
+            A list of resource contents (either TextResourceContents or BlobResourceContents).
+        """
+        async with self:  # Ensure server is running
+            result = await self._client.read_resource(AnyUrl(uri))
+        return result.contents
+
     async def __aenter__(self) -> Self:
         """Enter the MCP server context.
 

tests/mcp_server.py

@@ -124,6 +124,12 @@ async def product_name_resource() -> str:
     return Path(__file__).parent.joinpath('assets/product_name.txt').read_text()
 
 
+@mcp.resource('resource://greeting/{name}', mime_type='text/plain')
+async def greeting_resource_template(name: str) -> str:
+    """Dynamic greeting resource template."""
+    return f'Hello, {name}!'
+
+
 @mcp.tool()
 async def get_image() -> Image:
     data = Path(__file__).parent.joinpath('assets/kiwi.png').read_bytes()

tests/test_mcp.py

@@ -37,7 +37,15 @@
     from mcp import ErrorData, McpError, SamplingMessage
     from mcp.client.session import ClientSession
     from mcp.shared.context import RequestContext
-    from mcp.types import CreateMessageRequestParams, ElicitRequestParams, ElicitResult, ImageContent, TextContent
+    from mcp.types import (
+        BlobResourceContents,
+        CreateMessageRequestParams,
+        ElicitRequestParams,
+        ElicitResult,
+        ImageContent,
+        TextContent,
+        TextResourceContents,
+    )
 
     from pydantic_ai._mcp import map_from_mcp_params, map_from_model_response
     from pydantic_ai.mcp import CallToolFunc, MCPServerSSE, MCPServerStdio, ToolResult
@@ -318,6 +326,36 @@ async def test_log_level_unset(run_context: RunContext[int]):
         assert result == snapshot('unset')
 
 
+async def test_stdio_server_list_resources(run_context: RunContext[int]):
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+    async with server:
+        resources = await server.list_resources()
+        assert len(resources) == snapshot(3)
+
+        assert str(resources[0].uri) == snapshot('resource://kiwi.png')
+        assert resources[0].mimeType == snapshot('image/png')
+        assert resources[0].name == snapshot('kiwi_resource')
+
+        assert str(resources[1].uri) == snapshot('resource://marcelo.mp3')
+        assert resources[1].mimeType == snapshot('audio/mpeg')
+        assert resources[1].name == snapshot('marcelo_resource')
+
+        assert str(resources[2].uri) == snapshot('resource://product_name.txt')
+        assert resources[2].mimeType == snapshot('text/plain')
+        assert resources[2].name == snapshot('product_name_resource')
+
+
+async def test_stdio_server_list_resource_templates(run_context: RunContext[int]):
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+    async with server:
+        resource_templates = await server.list_resource_templates()
+        assert len(resource_templates) == snapshot(1)
+
+        assert resource_templates[0].uriTemplate == snapshot('resource://greeting/{name}')
+        assert resource_templates[0].name == snapshot('greeting_resource_template')
+        assert resource_templates[0].description == snapshot('Dynamic greeting resource template.')
+
+
 async def test_log_level_set(run_context: RunContext[int]):
     server = MCPServerStdio('python', ['-m', 'tests.mcp_server'], log_level='info')
     assert server.log_level == 'info'
@@ -1394,6 +1432,52 @@ async def test_elicitation_callback_not_set(run_context: RunContext[int]):
             await server.direct_call_tool('use_elicitation', {'question': 'Should I continue?'})
 
 
+async def test_read_text_resource(run_context: RunContext[int]):
+    """Test reading a text resource (TextResourceContents)."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+    async with server:
+        contents = await server.read_resource('resource://product_name.txt')
+        assert len(contents) == snapshot(1)
+
+        content = contents[0]
+        assert str(content.uri) == snapshot('resource://product_name.txt')
+        assert content.mimeType == snapshot('text/plain')
+        assert isinstance(content, TextResourceContents)
+        assert content.text == snapshot('Pydantic AI\n')
+
+
+async def test_read_blob_resource(run_context: RunContext[int]):
+    """Test reading a binary resource (BlobResourceContents)."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+    async with server:
+        contents = await server.read_resource('resource://kiwi.png')
+        assert len(contents) == snapshot(1)
+
+        content = contents[0]
+        assert str(content.uri) == snapshot('resource://kiwi.png')
+        assert content.mimeType == snapshot('image/png')
+        assert isinstance(content, BlobResourceContents)
+        # blob should be base64 encoded string
+        assert isinstance(content.blob, str)
+        # Decode and verify it's PNG data (starts with PNG magic bytes)
+        decoded_data = base64.b64decode(content.blob)
+        assert decoded_data[:8] == b'\x89PNG\r\n\x1a\n'  # PNG magic bytes
+
+
+async def test_read_resource_template(run_context: RunContext[int]):
+    """Test reading a resource template with parameters (TextResourceContents)."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+    async with server:
+        contents = await server.read_resource('resource://greeting/Alice')
+        assert len(contents) == snapshot(1)
+
+        content = contents[0]
+        assert str(content.uri) == snapshot('resource://greeting/Alice')
+        assert content.mimeType == snapshot('text/plain')
+        assert isinstance(content, TextResourceContents)
+        assert content.text == snapshot('Hello, Alice!')
+
+
 def test_load_mcp_servers(tmp_path: Path):
     config = tmp_path / 'mcp.json'