Instantiate MCP resource related Resource dataclasses in a consistent way.

Author: fennb

pydantic_ai_slim/pydantic_ai/_mcp.py

@@ -2,7 +2,7 @@
 
 import base64
 from collections.abc import Sequence
-from typing import TYPE_CHECKING, Literal
+from typing import Literal
 
 from . import exceptions, messages
 
@@ -14,9 +14,6 @@
         'you can use the `mcp` optional group — `pip install "pydantic-ai-slim[mcp]"`'
     ) from _import_error
 
-if TYPE_CHECKING:
-    from .mcp import Resource, ResourceTemplate, ServerCapabilities
-
 
 def map_from_mcp_params(params: mcp_types.CreateMessageRequestParams) -> list[messages.ModelMessage]:
     """Convert from MCP create message request parameters to pydantic-ai messages."""
@@ -126,56 +123,3 @@ def map_from_sampling_content(
         return messages.TextPart(content=content.text)
     else:
         raise NotImplementedError('Image and Audio responses in sampling are not yet supported')
-
-
-def map_from_mcp_resource(mcp_resource: mcp_types.Resource) -> Resource:
-    """Convert from MCP Resource to native Pydantic AI Resource."""
-    from .mcp import Resource, ResourceAnnotations
-
-    return Resource(
-        uri=str(mcp_resource.uri),
-        name=mcp_resource.name,
-        title=mcp_resource.title,
-        description=mcp_resource.description,
-        mime_type=mcp_resource.mimeType,
-        size=mcp_resource.size,
-        annotations=(
-            ResourceAnnotations(audience=mcp_resource.annotations.audience, priority=mcp_resource.annotations.priority)
-            if mcp_resource.annotations
-            else None
-        ),
-        metadata=mcp_resource.meta,
-    )
-
-
-def map_from_mcp_resource_template(mcp_template: mcp_types.ResourceTemplate) -> ResourceTemplate:
-    """Convert from MCP ResourceTemplate to native Pydantic AI ResourceTemplate."""
-    from .mcp import ResourceAnnotations, ResourceTemplate
-
-    return ResourceTemplate(
-        uri_template=mcp_template.uriTemplate,
-        name=mcp_template.name,
-        title=mcp_template.title,
-        description=mcp_template.description,
-        mime_type=mcp_template.mimeType,
-        annotations=(
-            ResourceAnnotations(audience=mcp_template.annotations.audience, priority=mcp_template.annotations.priority)
-            if mcp_template.annotations
-            else None
-        ),
-        metadata=mcp_template.meta,
-    )
-
-
-def map_from_mcp_server_capabilities(mcp_capabilities: mcp_types.ServerCapabilities) -> ServerCapabilities:
-    """Convert from MCP ServerCapabilities to native Pydantic AI ServerCapabilities."""
-    from .mcp import ServerCapabilities
-
-    return ServerCapabilities(
-        experimental=list(mcp_capabilities.experimental.keys()) if mcp_capabilities.experimental else None,
-        logging=mcp_capabilities.logging is not None,
-        prompts=mcp_capabilities.prompts is not None,
-        resources=mcp_capabilities.resources is not None,
-        tools=mcp_capabilities.tools is not None,
-        completions=mcp_capabilities.completions is not None,
-    )

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -114,6 +114,15 @@ class ResourceAnnotations:
 
     __repr__ = _utils.dataclasses_no_defaults_repr
 
+    @classmethod
+    def from_mcp_sdk(cls, mcp_annotations: mcp_types.Annotations) -> ResourceAnnotations:
+        """Convert from MCP SDK Annotations to ResourceAnnotations.
+
+        Args:
+            mcp_annotations: The MCP SDK annotations object.
+        """
+        return cls(audience=mcp_annotations.audience, priority=mcp_annotations.priority)
+
 
 @dataclass(repr=False, kw_only=True)
 class BaseResource(ABC):
@@ -150,6 +159,26 @@ class Resource(BaseResource):
     size: int | None = None
     """The size of the raw resource content in bytes (before base64 encoding), if known."""
 
+    @classmethod
+    def from_mcp_sdk(cls, mcp_resource: mcp_types.Resource) -> Resource:
+        """Convert from MCP SDK Resource to PydanticAI Resource.
+
+        Args:
+            mcp_resource: The MCP SDK Resource object.
+        """
+        return cls(
+            uri=str(mcp_resource.uri),
+            name=mcp_resource.name,
+            title=mcp_resource.title,
+            description=mcp_resource.description,
+            mime_type=mcp_resource.mimeType,
+            size=mcp_resource.size,
+            annotations=ResourceAnnotations.from_mcp_sdk(mcp_resource.annotations)
+            if mcp_resource.annotations
+            else None,
+            metadata=mcp_resource.meta,
+        )
+
 
 @dataclass(repr=False, kw_only=True)
 class ResourceTemplate(BaseResource):
@@ -158,6 +187,25 @@ class ResourceTemplate(BaseResource):
     uri_template: str
     """URI template (RFC 6570) for constructing resource URIs."""
 
+    @classmethod
+    def from_mcp_sdk(cls, mcp_template: mcp_types.ResourceTemplate) -> ResourceTemplate:
+        """Convert from MCP SDK ResourceTemplate to PydanticAI ResourceTemplate.
+
+        Args:
+            mcp_template: The MCP SDK ResourceTemplate object.
+        """
+        return cls(
+            uri_template=mcp_template.uriTemplate,
+            name=mcp_template.name,
+            title=mcp_template.title,
+            description=mcp_template.description,
+            mime_type=mcp_template.mimeType,
+            annotations=ResourceAnnotations.from_mcp_sdk(mcp_template.annotations)
+            if mcp_template.annotations
+            else None,
+            metadata=mcp_template.meta,
+        )
+
 
 @dataclass(repr=False, kw_only=True)
 class ServerCapabilities:
@@ -183,6 +231,22 @@ class ServerCapabilities:
 
     __repr__ = _utils.dataclasses_no_defaults_repr
 
+    @classmethod
+    def from_mcp_sdk(cls, mcp_capabilities: mcp_types.ServerCapabilities) -> ServerCapabilities:
+        """Convert from MCP SDK ServerCapabilities to PydanticAI ServerCapabilities.
+
+        Args:
+            mcp_capabilities: The MCP SDK ServerCapabilities object.
+        """
+        return cls(
+            experimental=list(mcp_capabilities.experimental.keys()) if mcp_capabilities.experimental else None,
+            logging=mcp_capabilities.logging is not None,
+            prompts=mcp_capabilities.prompts is not None,
+            resources=mcp_capabilities.resources is not None,
+            tools=mcp_capabilities.tools is not None,
+            completions=mcp_capabilities.completions is not None,
+        )
+
 
 TOOL_SCHEMA_VALIDATOR = pydantic_core.SchemaValidator(
     schema=pydantic_core.core_schema.dict_schema(
@@ -486,7 +550,7 @@ async def list_resources(self) -> list[Resource]:
                 result = await self._client.list_resources()
             except mcp_exceptions.McpError as e:
                 raise MCPError.from_mcp_sdk_error(e) from e
-        return [_mcp.map_from_mcp_resource(r) for r in result.resources]
+        return [Resource.from_mcp_sdk(r) for r in result.resources]
 
     async def list_resource_templates(self) -> list[ResourceTemplate]:
         """Retrieve resource templates that are currently present on the server.
@@ -501,7 +565,7 @@ async def list_resource_templates(self) -> list[ResourceTemplate]:
                 result = await self._client.list_resource_templates()
             except mcp_exceptions.McpError as e:
                 raise MCPError.from_mcp_sdk_error(e) from e
-        return [_mcp.map_from_mcp_resource_template(t) for t in result.resourceTemplates]
+        return [ResourceTemplate.from_mcp_sdk(t) for t in result.resourceTemplates]
 
     @overload
     async def read_resource(self, uri: str) -> str | messages.BinaryContent | list[str | messages.BinaryContent]: ...
@@ -564,7 +628,7 @@ async def __aenter__(self) -> Self:
                     with anyio.fail_after(self.timeout):
                         result = await self._client.initialize()
                         self._server_info = result.serverInfo
-                        self._server_capabilities = _mcp.map_from_mcp_server_capabilities(result.capabilities)
+                        self._server_capabilities = ServerCapabilities.from_mcp_sdk(result.capabilities)
                         self._instructions = result.instructions
                         if log_level := self.log_level:
                             await self._client.set_logging_level(log_level)