From ccaf9b39661bd5de5af5c2df7335871456c12c8b Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 15 Apr 2025 10:23:00 -0400 Subject: [PATCH] Generate MCP resources and templates on respective classes --- src/fastmcp/resources/resource.py | 21 +++++++++++++++++- src/fastmcp/resources/resource_manager.py | 24 +++++++++++++++++++-- src/fastmcp/resources/template.py | 20 +++++++++++++++++ src/fastmcp/server/server.py | 26 ++++++----------------- 4 files changed, 69 insertions(+), 22 deletions(-) diff --git a/src/fastmcp/resources/resource.py b/src/fastmcp/resources/resource.py index a583d6357..b79381bbd 100644 --- a/src/fastmcp/resources/resource.py +++ b/src/fastmcp/resources/resource.py @@ -1,8 +1,9 @@ """Base classes and interfaces for FastMCP resources.""" import abc -from typing import Annotated +from typing import Annotated, Any +from mcp.types import Resource as MCPResource from pydantic import ( AnyUrl, BaseModel, @@ -38,6 +39,14 @@ class Resource(BaseModel, abc.ABC): pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$", ) + @field_validator("mime_type", mode="before") + @classmethod + def set_default_mime_type(cls, mime_type: str | None) -> str: + """Set default MIME type if not provided.""" + if mime_type: + return mime_type + return "text/plain" + @field_validator("name", mode="before") @classmethod def set_default_name(cls, name: str | None, info: ValidationInfo) -> str: @@ -57,3 +66,13 @@ class Resource(BaseModel, abc.ABC): if not isinstance(other, Resource): return False return self.model_dump() == other.model_dump() + + def to_mcp_resource(self, **overrides: Any) -> MCPResource: + """Convert the resource to an MCPResource.""" + kwargs = { + "uri": self.uri, + "name": self.name, + "description": self.description, + "mimeType": self.mime_type, + } + return MCPResource(**kwargs | overrides) diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py index 1dc21a45e..ed6f18f66 100644 --- a/src/fastmcp/resources/resource_manager.py +++ b/src/fastmcp/resources/resource_manager.py @@ -7,8 +7,13 @@ from typing import Any from pydantic import AnyUrl from fastmcp.exceptions import ResourceError -from fastmcp.resources import FunctionResource, Resource -from fastmcp.resources.template import ResourceTemplate, match_uri_template +from fastmcp.resources import FunctionResource +from fastmcp.resources.resource import MCPResource, Resource +from fastmcp.resources.template import ( + MCPResourceTemplate, + ResourceTemplate, + match_uri_template, +) from fastmcp.settings import DuplicateBehavior from fastmcp.utilities.logging import get_logger @@ -227,6 +232,21 @@ class ResourceManager: logger.debug("Listing resources", extra={"count": len(self._resources)}) return list(self._resources.values()) + def list_mcp_resources(self) -> list[MCPResource]: + """List all registered resources in the format expected by the low-level MCP server.""" + + return [ + resource.to_mcp_resource(uri=key) + for key, resource in self._resources.items() + ] + + def list_mcp_resource_templates(self) -> list[MCPResourceTemplate]: + """List all registered resource templates in the format expected by the low-level MCP server.""" + return [ + template.to_mcp_template(uriTemplate=key) + for key, template in self._templates.items() + ] + def get_templates(self) -> dict[str, ResourceTemplate]: """Get all registered templates, keyed by URI template.""" return self._templates diff --git a/src/fastmcp/resources/template.py b/src/fastmcp/resources/template.py index 7d05aa057..bb92b15d8 100644 --- a/src/fastmcp/resources/template.py +++ b/src/fastmcp/resources/template.py @@ -7,12 +7,14 @@ import re from collections.abc import Callable from typing import Annotated, Any +from mcp.types import ResourceTemplate as MCPResourceTemplate from pydantic import ( AnyUrl, BaseModel, BeforeValidator, Field, TypeAdapter, + field_validator, validate_call, ) @@ -62,6 +64,14 @@ class ResourceTemplate(BaseModel): description="JSON schema for function parameters" ) + @field_validator("mime_type", mode="before") + @classmethod + def set_default_mime_type(cls, mime_type: str | None) -> str: + """Set default MIME type if not provided.""" + if mime_type: + return mime_type + return "text/plain" + @classmethod def from_function( cls, @@ -144,3 +154,13 @@ class ResourceTemplate(BaseModel): if not isinstance(other, ResourceTemplate): return False return self.model_dump() == other.model_dump() + + def to_mcp_template(self, **overrides: Any) -> MCPResourceTemplate: + """Convert the resource template to an MCPResourceTemplate.""" + kwargs = { + "uriTemplate": self.uri_template, + "name": self.name, + "description": self.description, + "mimeType": self.mime_type, + } + return MCPResourceTemplate(**kwargs | overrides) diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index c53e1ee4e..b99e5d867 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -225,7 +225,12 @@ class FastMCP(Generic[LifespanResultT]): converted_result = _convert_to_content(result) return converted_result + def get_resources(self) -> dict[str, Resource]: + """Get all registered resources, indexed by registered key.""" + return self._resource_manager.get_resources() + def list_resources(self) -> list[Resource]: + """List all registered resources.""" return self._resource_manager.list_resources() async def _mcp_list_resources(self) -> list[MCPResource]: @@ -236,16 +241,7 @@ class FastMCP(Generic[LifespanResultT]): See `list_resources` for a more ergonomic way to list resources. """ - resources = self.list_resources() - return [ - MCPResource( - uri=resource.uri, - name=resource.name or "", - description=resource.description, - mimeType=resource.mime_type, - ) - for resource in resources - ] + return self._resource_manager.list_mcp_resources() def list_resource_templates(self) -> list[ResourceTemplate]: return self._resource_manager.list_templates() @@ -258,15 +254,7 @@ class FastMCP(Generic[LifespanResultT]): See `list_resource_templates` for a more ergonomic way to list resource templates. """ - templates = self.list_resource_templates() - return [ - MCPResourceTemplate( - uriTemplate=template.uri_template, - name=template.name, - description=template.description, - ) - for template in templates - ] + return self._resource_manager.list_mcp_resource_templates() async def read_resource(self, uri: AnyUrl | str) -> str | bytes: """Read a resource by URI."""