Generate MCP resources and templates on respective classes

This commit is contained in:
Jeremiah Lowin 2025-04-15 10:23:00 -04:00
commit ccaf9b3966
4 changed files with 69 additions and 22 deletions

View file

@ -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)

View file

@ -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

View file

@ -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)

View file

@ -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."""