Refactor resources module

src/fastmcp/resources/__init__.py

@@ -0,0 +1,23 @@
+from .base import Resource
+from .types import (
+    TextResource,
+    BinaryResource,
+    FunctionResource,
+    FileResource,
+    HttpResource,
+    DirectoryResource,
+)
+from .templates import ResourceTemplate
+from .manager import ResourceManager
+
+__all__ = [
+    "Resource",
+    "TextResource",
+    "BinaryResource",
+    "FunctionResource",
+    "FileResource",
+    "HttpResource",
+    "DirectoryResource",
+    "ResourceTemplate",
+    "ResourceManager",
+]

src/fastmcp/resources/base.py

@@ -0,0 +1,47 @@
+"""Base classes and interfaces for FastMCP resources."""
+
+import abc
+from typing import Union
+
+from pydantic import BaseModel, Field, field_validator
+from pydantic.networks import _BaseUrl
+
+
+class Resource(BaseModel, abc.ABC):
+    """Base class for all resources."""
+
+    uri: _BaseUrl = Field(description="URI of the resource")
+    name: str = Field(description="Name of the resource", default=None)
+    description: str | None = Field(
+        description="Description of the resource", default=None
+    )
+    mime_type: str = Field(
+        default="text/plain",
+        description="MIME type of the resource content",
+        pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
+    )
+
+    @field_validator("name", mode="before")
+    @classmethod
+    def set_default_name(cls, name: str | None, info) -> str:
+        """Set default name from URI if not provided."""
+        if name:
+            return name
+        # Extract everything after the protocol (e.g., "desktop" from "resource://desktop")
+        uri = info.data.get("uri")
+        if uri:
+            uri_str = str(uri)
+            if "://" in uri_str:
+                name = uri_str.split("://", 1)[1]
+                if name:
+                    return name
+        raise ValueError("Either name or uri must be provided")
+
+    @abc.abstractmethod
+    async def read(self) -> Union[str, bytes]:
+        """Read the resource content."""
+        pass
+
+    model_config = {
+        "validate_default": True,
+    }

src/fastmcp/resources/manager.py

@@ -0,0 +1,89 @@
+"""Resource manager functionality."""
+
+from typing import Callable, Dict, Optional, Union
+
+from pydantic.networks import _BaseUrl
+
+from fastmcp.resources.base import Resource
+from fastmcp.resources.templates import ResourceTemplate
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ResourceManager:
+    """Manages FastMCP resources."""
+
+    def __init__(self, warn_on_duplicate_resources: bool = True):
+        self._resources: Dict[str, Resource] = {}
+        self._templates: Dict[str, ResourceTemplate] = {}
+        self.warn_on_duplicate_resources = warn_on_duplicate_resources
+
+    def add_template(
+        self,
+        func: Callable,
+        uri_template: str,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        mime_type: Optional[str] = None,
+    ) -> ResourceTemplate:
+        """Add a template from a function."""
+        template = ResourceTemplate.from_function(
+            func,
+            uri_template=uri_template,
+            name=name,
+            description=description,
+            mime_type=mime_type,
+        )
+        self._templates[template.uri_template] = template
+        return template
+
+    async def get_resource(self, uri: Union[_BaseUrl, str]) -> Optional[Resource]:
+        """Get resource by URI, checking concrete resources first, then templates."""
+        uri_str = str(uri)
+        logger.debug("Getting resource", extra={"uri": uri_str})
+
+        # First check concrete resources
+        if resource := self._resources.get(uri_str):
+            return resource
+
+        # Then check templates
+        for template in self._templates.values():
+            if params := template.matches(uri_str):
+                try:
+                    return await template.create_resource(uri_str, params)
+                except Exception as e:
+                    raise ValueError(f"Error creating resource from template: {e}")
+
+        raise ValueError(f"Unknown resource: {uri}")
+
+    def list_resources(self) -> list[Resource]:
+        """List all registered resources."""
+        logger.debug("Listing resources", extra={"count": len(self._resources)})
+        return list(self._resources.values())
+
+    def add_resource(self, resource: Resource) -> Resource:
+        """Add a resource to the manager.
+
+        Args:
+            resource: A Resource instance to add
+
+        Returns:
+            The added resource. If a resource with the same URI already exists,
+            returns the existing resource.
+        """
+        logger.debug(
+            "Adding resource",
+            extra={
+                "uri": resource.uri,
+                "type": type(resource).__name__,
+                "name": resource.name,
+            },
+        )
+        existing = self._resources.get(str(resource.uri))
+        if existing:
+            if self.warn_on_duplicate_resources:
+                logger.warning(f"Resource already exists: {resource.uri}")
+            return existing
+        self._resources[str(resource.uri)] = resource
+        return resource

src/fastmcp/resources/templates.py

@@ -0,0 +1,80 @@
+"""Resource template functionality."""
+
+import inspect
+import re
+from typing import Any, Callable, Dict, Optional
+
+from pydantic import BaseModel, Field, TypeAdapter, validate_call
+
+from fastmcp.resources.types import FunctionResource, Resource
+
+
+class ResourceTemplate(BaseModel):
+    """A template for dynamically creating resources."""
+
+    uri_template: str = Field(
+        description="URI template with parameters (e.g. weather://{city}/current)"
+    )
+    name: str = Field(description="Name of the resource")
+    description: str | None = Field(description="Description of what the resource does")
+    mime_type: str = Field(
+        default="text/plain", description="MIME type of the resource content"
+    )
+    func: Callable = Field(exclude=True)
+    parameters: dict = Field(description="JSON schema for function parameters")
+
+    @classmethod
+    def from_function(
+        cls,
+        func: Callable,
+        uri_template: str,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        mime_type: Optional[str] = None,
+    ) -> "ResourceTemplate":
+        """Create a template from a function."""
+        func_name = name or func.__name__
+        if func_name == "<lambda>":
+            raise ValueError("You must provide a name for lambda functions")
+
+        # Get schema from TypeAdapter - will fail if function isn't properly typed
+        parameters = TypeAdapter(func).json_schema()
+
+        # ensure the arguments are properly cast
+        func = validate_call(func)
+
+        return cls(
+            uri_template=uri_template,
+            name=func_name,
+            description=description or func.__doc__ or "",
+            mime_type=mime_type or "text/plain",
+            func=func,
+            parameters=parameters,
+        )
+
+    def matches(self, uri: str) -> Optional[Dict[str, Any]]:
+        """Check if URI matches template and extract parameters."""
+        # Convert template to regex pattern
+        pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
+        match = re.match(f"^{pattern}$", uri)
+        if match:
+            return match.groupdict()
+        return None
+
+    async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource:
+        """Create a resource from the template with the given parameters."""
+        try:
+            # Call function and check if result is a coroutine
+            result = self.func(**params)
+            if inspect.iscoroutine(result):
+                result = await result
+
+            return FunctionResource(
+                uri=uri,
+                name=self.name,
+                description=self.description,
+                mime_type=self.mime_type,
+                func=lambda: result,  # Capture result in closure
+            )
+        except Exception as e:
+            raise ValueError(f"Error creating resource from template: {e}")

src/fastmcp/{resources.py → resources/types.py}

@@ -1,59 +1,15 @@
-import inspect
-import pydantic.json
-import abc
+"""Concrete resource implementations."""
+
 import asyncio
 import json
-import re
 from pathlib import Path
-from typing import Dict, Optional, Callable, Any, Union
+from typing import Any, Callable, Union
 
 import httpx
-from pydantic import BaseModel, Field, TypeAdapter, validate_call, field_validator
-from pydantic.networks import _BaseUrl
-
-from .utilities.logging import get_logger
-
-logger = get_logger(__name__)
-
-
-class Resource(BaseModel, abc.ABC):
-    """Base class for all resources."""
-
-    uri: _BaseUrl = Field(description="URI of the resource")
-    name: str = Field(description="Name of the resource", default=None)
-    description: Optional[str] = Field(
-        description="Description of the resource", default=None
-    )
-    mime_type: str = Field(
-        default="text/plain",
-        description="MIME type of the resource content",
-        pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
-    )
-
-    @field_validator("name", mode="before")
-    @classmethod
-    def set_default_name(cls, name: str | None, info) -> str:
-        """Set default name from URI if not provided."""
-        if name:
-            return name
-        # Extract everything after the protocol (e.g., "desktop" from "resource://desktop")
-        uri = info.data.get("uri")
-        if uri:
-            uri_str = str(uri)
-            if "://" in uri_str:
-                name = uri_str.split("://", 1)[1]
-                if name:
-                    return name
-        raise ValueError("Either name or uri must be provided")
-
-    @abc.abstractmethod
-    async def read(self) -> Union[str, bytes]:
-        """Read the resource content."""
-        pass
+import pydantic.json
+from pydantic import Field
 
-    model_config = {
-        "validate_default": True,
-    }
+from fastmcp.resources.base import Resource
 
 
 class TextResource(Resource):
@@ -126,7 +82,7 @@ class FileResource(Resource):
         description="MIME type of the resource content",
     )
 
-    @field_validator("path")
+    @pydantic.field_validator("path")
     @classmethod
     def validate_absolute_path(cls, path: Path) -> Path:
         """Ensure path is absolute."""
@@ -148,7 +104,7 @@ class HttpResource(Resource):
     """A resource that reads from an HTTP endpoint."""
 
     url: str = Field(description="URL to fetch content from")
-    mime_type: Optional[str] = Field(
+    mime_type: str | None = Field(
         default="application/json", description="MIME type of the resource content"
     )
 
@@ -167,14 +123,14 @@ class DirectoryResource(Resource):
     recursive: bool = Field(
         default=False, description="Whether to list files recursively"
     )
-    pattern: Optional[str] = Field(
+    pattern: str | None = Field(
         default=None, description="Optional glob pattern to filter files"
     )
-    mime_type: Optional[str] = Field(
+    mime_type: str | None = Field(
         default="application/json", description="MIME type of the resource content"
     )
 
-    @field_validator("path")
+    @pydantic.field_validator("path")
     @classmethod
     def validate_absolute_path(cls, path: Path) -> Path:
         """Ensure path is absolute."""
@@ -212,152 +168,3 @@ class DirectoryResource(Resource):
             return json.dumps({"files": file_list}, indent=2)
         except Exception as e:
             raise ValueError(f"Error reading directory {self.path}: {e}")
-
-
-class ResourceTemplate(BaseModel):
-    """A template for dynamically creating resources."""
-
-    uri_template: str = Field(
-        description="URI template with parameters (e.g. weather://{city}/current)"
-    )
-    name: str = Field(description="Name of the resource")
-    description: Optional[str] = Field(
-        description="Description of what the resource does"
-    )
-    mime_type: str = Field(
-        default="text/plain", description="MIME type of the resource content"
-    )
-    func: Callable = Field(exclude=True)
-    parameters: dict = Field(description="JSON schema for function parameters")
-
-    @classmethod
-    def from_function(
-        cls,
-        func: Callable,
-        uri_template: str,
-        name: Optional[str] = None,
-        description: Optional[str] = None,
-        mime_type: Optional[str] = None,
-    ) -> "ResourceTemplate":
-        """Create a template from a function."""
-        func_name = name or func.__name__
-        if func_name == "<lambda>":
-            raise ValueError("You must provide a name for lambda functions")
-
-        # Get schema from TypeAdapter - will fail if function isn't properly typed
-        parameters = TypeAdapter(func).json_schema()
-
-        # ensure the arguments are properly cast
-        func = validate_call(func)
-
-        return cls(
-            uri_template=uri_template,
-            name=func_name,
-            description=description or func.__doc__ or "",
-            func=func,
-            parameters=parameters,
-        )
-
-    def matches(self, uri: str) -> Optional[Dict[str, Any]]:
-        """Check if URI matches template and extract parameters."""
-        # Convert template to regex pattern
-        pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
-        match = re.match(f"^{pattern}$", uri)
-        if match:
-            return match.groupdict()
-        return None
-
-    async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource:
-        """Create a resource from the template with the given parameters."""
-        try:
-            # Call function and check if result is a coroutine
-            result = self.func(**params)
-            if inspect.iscoroutine(result):
-                result = await result
-
-            return FunctionResource(
-                uri=uri,
-                name=self.name,
-                description=self.description,
-                func=lambda: result,  # Capture result in closure
-            )
-        except Exception as e:
-            raise ValueError(f"Error creating resource from template: {e}")
-
-
-class ResourceManager:
-    """Manages FastMCP resources."""
-
-    def __init__(self, warn_on_duplicate_resources: bool = True):
-        self._resources: Dict[str, Resource] = {}
-        self._templates: Dict[str, ResourceTemplate] = {}
-        self.warn_on_duplicate_resources = warn_on_duplicate_resources
-
-    def add_template(
-        self,
-        func: Callable,
-        uri_template: str,
-        name: Optional[str] = None,
-        description: Optional[str] = None,
-        mime_type: Optional[str] = None,
-    ) -> ResourceTemplate:
-        """Add a template from a function."""
-        template = ResourceTemplate.from_function(
-            func,
-            uri_template=uri_template,
-            name=name,
-            description=description,
-            mime_type=mime_type,
-        )
-        self._templates[template.uri_template] = template
-        return template
-
-    async def get_resource(self, uri: Union[_BaseUrl, str]) -> Optional[Resource]:
-        """Get resource by URI, checking concrete resources first, then templates."""
-        uri_str = str(uri)
-        logger.debug("Getting resource", extra={"uri": uri_str})
-
-        # First check concrete resources
-        if resource := self._resources.get(uri_str):
-            return resource
-
-        # Then check templates
-        for template in self._templates.values():
-            if params := template.matches(uri_str):
-                try:
-                    return await template.create_resource(uri_str, params)
-                except Exception as e:
-                    raise ValueError(f"Error creating resource from template: {e}")
-
-        raise ValueError(f"Unknown resource: {uri}")
-
-    def list_resources(self) -> list[Resource]:
-        """List all registered resources."""
-        logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return list(self._resources.values())
-
-    def add_resource(self, resource: Resource) -> Resource:
-        """Add a resource to the manager.
-
-        Args:
-            resource: A Resource instance to add
-
-        Returns:
-            The added resource. If a resource with the same URI already exists,
-            returns the existing resource.
-        """
-        logger.debug(
-            "Adding resource",
-            extra={
-                "uri": resource.uri,
-                "type": type(resource).__name__,
-                "name": resource.name,
-            },
-        )
-        existing = self._resources.get(str(resource.uri))
-        if existing:
-            if self.warn_on_duplicate_resources:
-                logger.warning(f"Resource already exists: {resource.uri}")
-            return existing
-        self._resources[str(resource.uri)] = resource
-        return resource

src/fastmcp/server.py

@@ -20,14 +20,11 @@ from mcp.types import (
 from pydantic_settings import BaseSettings
 from pydantic.networks import _BaseUrl
 
-from .exceptions import ResourceError
-from .resources import (
-    Resource,
-    FunctionResource,
-    ResourceManager,
-)
-from .tools import ToolManager, Image
-from .utilities.logging import get_logger, configure_logging
+from fastmcp.exceptions import ResourceError
+from fastmcp.resources import Resource, ResourceManager
+from fastmcp.resources.types import FunctionResource
+from fastmcp.tools import ToolManager, Image
+from fastmcp.utilities.logging import get_logger, configure_logging
 
 logger = get_logger(__name__)