Introduce native MCPError type and use it in MCP server resource methods.

Author: fennb

pydantic_ai_slim/pydantic_ai/exceptions.py

@@ -23,6 +23,8 @@
     'UnexpectedModelBehavior',
     'UsageLimitExceeded',
     'ModelHTTPError',
+    'MCPError',
+    'ServerCapabilitiesError',
     'FallbackExceptionGroup',
 )
 
@@ -158,6 +160,24 @@ def __init__(self, status_code: int, model_name: str, body: object | None = None
         super().__init__(message)
 
 
+class MCPError(RuntimeError):
+    """Base class for errors occurring during interaction with an MCP server."""
+
+    message: str
+    """The error message."""
+
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(message)
+
+    def __str__(self) -> str:
+        return self.message
+
+
+class ServerCapabilitiesError(MCPError):
+    """Raised when attempting to access server capabilities that aren't present."""
+
+
 class FallbackExceptionGroup(ExceptionGroup):
     """A group of exceptions that can be raised when all fallback models fail."""
 

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -319,14 +319,29 @@ async def list_resources(self) -> list[_mcp.Resource]:
         Note:
         - We don't cache resources as they might change.
         - We also don't subscribe to resource changes to avoid complexity.
+
+        Raises:
+            ServerCapabilitiesError: If the server does not support resources.
         """
         async with self:  # Ensure server is running
+            if not self.capabilities.resources:
+                raise exceptions.ServerCapabilitiesError(
+                    f'Server does not support resources capability. Available capabilities: {self.capabilities}'
+                )
             result = await self._client.list_resources()
         return [_mcp.map_from_mcp_resource(r) for r in result.resources]
 
     async def list_resource_templates(self) -> list[_mcp.ResourceTemplate]:
-        """Retrieve resource templates that are currently present on the server."""
+        """Retrieve resource templates that are currently present on the server.
+
+        Raises:
+            ServerCapabilitiesError: If the server does not support resources.
+        """
         async with self:  # Ensure server is running
+            if not self.capabilities.resources:
+                raise exceptions.ServerCapabilitiesError(
+                    f'Server does not support resources capability. Available capabilities: {self.capabilities}'
+                )
             result = await self._client.list_resource_templates()
         return [_mcp.map_from_mcp_resource_template(t) for t in result.resourceTemplates]
 
@@ -349,9 +364,16 @@ async def read_resource(
         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.
+
+        Raises:
+            ServerCapabilitiesError: If the server does not support resources.
         """
         resource_uri = uri if isinstance(uri, str) else uri.uri
         async with self:  # Ensure server is running
+            if not self.capabilities.resources:
+                raise exceptions.ServerCapabilitiesError(
+                    f'Server does not support resources capability. Available capabilities: {self.capabilities}'
+                )
             result = await self._client.read_resource(AnyUrl(resource_uri))
         return (
             self._get_content(result.contents[0])

tests/test_mcp.py

@@ -23,9 +23,9 @@
     ToolReturnPart,
     UserPromptPart,
 )
-from pydantic_ai._mcp import Resource
+from pydantic_ai._mcp import Resource, ServerCapabilities
 from pydantic_ai.agent import Agent
-from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior, UserError
+from pydantic_ai.exceptions import ModelRetry, ServerCapabilitiesError, UnexpectedModelBehavior, UserError
 from pydantic_ai.mcp import MCPServerStreamableHTTP, load_mcp_servers
 from pydantic_ai.models import Model
 from pydantic_ai.models.test import TestModel
@@ -1583,3 +1583,19 @@ async def test_capabilities(mcp_server: MCPServerStdio) -> None:
         assert mcp_server.capabilities.logging is True
         assert mcp_server.capabilities.completions is False
         assert mcp_server.capabilities.experimental is None
+
+
+async def test_resource_methods_without_capability(mcp_server: MCPServerStdio) -> None:
+    """Test that resource methods raise ServerCapabilitiesError when resources capability is not available."""
+    async with mcp_server:
+        # Mock the capabilities to not support resources
+        mock_capabilities = ServerCapabilities(resources=False)
+        with patch.object(mcp_server, '_server_capabilities', mock_capabilities):
+            with pytest.raises(ServerCapabilitiesError, match='Server does not support resources capability'):
+                await mcp_server.list_resources()
+
+            with pytest.raises(ServerCapabilitiesError, match='Server does not support resources capability'):
+                await mcp_server.list_resource_templates()
+
+            with pytest.raises(ServerCapabilitiesError, match='Server does not support resources capability'):
+                await mcp_server.read_resource('resource://test')