mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-20 20:44:17 +02:00
feat: Add Annotations support for resources and resource templates (#1260)
Co-authored-by: Tapan Chugh <tapanc@cs.washington.edu>
This commit is contained in:
parent
d043511370
commit
51ceb23291
4 changed files with 78 additions and 0 deletions
|
|
@ -8,6 +8,7 @@ from collections.abc import Callable
|
|||
from typing import TYPE_CHECKING, Annotated, Any
|
||||
|
||||
import pydantic_core
|
||||
from mcp.types import Annotations
|
||||
from mcp.types import Resource as MCPResource
|
||||
from pydantic import (
|
||||
AnyUrl,
|
||||
|
|
@ -43,6 +44,10 @@ class Resource(FastMCPComponent, abc.ABC):
|
|||
description="MIME type of the resource content",
|
||||
pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
|
||||
)
|
||||
annotations: Annotated[
|
||||
Annotations | None,
|
||||
Field(description="Optional annotations about the resource's behavior"),
|
||||
] = None
|
||||
|
||||
def enable(self) -> None:
|
||||
super().enable()
|
||||
|
|
@ -70,6 +75,7 @@ class Resource(FastMCPComponent, abc.ABC):
|
|||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
annotations: Annotations | None = None,
|
||||
) -> FunctionResource:
|
||||
return FunctionResource.from_function(
|
||||
fn=fn,
|
||||
|
|
@ -80,6 +86,7 @@ class Resource(FastMCPComponent, abc.ABC):
|
|||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
annotations=annotations,
|
||||
)
|
||||
|
||||
@field_validator("mime_type", mode="before")
|
||||
|
|
@ -114,6 +121,7 @@ class Resource(FastMCPComponent, abc.ABC):
|
|||
"description": self.description,
|
||||
"mimeType": self.mime_type,
|
||||
"title": self.title,
|
||||
"annotations": self.annotations,
|
||||
}
|
||||
return MCPResource(**kwargs | overrides)
|
||||
|
||||
|
|
@ -157,6 +165,7 @@ class FunctionResource(Resource):
|
|||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
annotations: Annotations | None = None,
|
||||
) -> FunctionResource:
|
||||
"""Create a FunctionResource from a function."""
|
||||
if isinstance(uri, str):
|
||||
|
|
@ -170,6 +179,7 @@ class FunctionResource(Resource):
|
|||
mime_type=mime_type or "text/plain",
|
||||
tags=tags or set(),
|
||||
enabled=enabled if enabled is not None else True,
|
||||
annotations=annotations,
|
||||
)
|
||||
|
||||
async def read(self) -> str | bytes:
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from collections.abc import Callable
|
|||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
from mcp.types import Annotations
|
||||
from mcp.types import ResourceTemplate as MCPResourceTemplate
|
||||
from pydantic import (
|
||||
Field,
|
||||
|
|
@ -61,6 +62,9 @@ class ResourceTemplate(FastMCPComponent):
|
|||
parameters: dict[str, Any] = Field(
|
||||
description="JSON schema for function parameters"
|
||||
)
|
||||
annotations: Annotations | None = Field(
|
||||
default=None, description="Optional annotations about the resource's behavior"
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}(uri_template={self.uri_template!r}, name={self.name!r}, description={self.description!r}, tags={self.tags})"
|
||||
|
|
@ -91,6 +95,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
annotations: Annotations | None = None,
|
||||
) -> FunctionResourceTemplate:
|
||||
return FunctionResourceTemplate.from_function(
|
||||
fn=fn,
|
||||
|
|
@ -101,6 +106,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
annotations=annotations,
|
||||
)
|
||||
|
||||
@field_validator("mime_type", mode="before")
|
||||
|
|
@ -147,6 +153,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
"description": self.description,
|
||||
"mimeType": self.mime_type,
|
||||
"title": self.title,
|
||||
"annotations": self.annotations,
|
||||
}
|
||||
return MCPResourceTemplate(**kwargs | overrides)
|
||||
|
||||
|
|
@ -205,6 +212,7 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
annotations: Annotations | None = None,
|
||||
) -> FunctionResourceTemplate:
|
||||
"""Create a template from a function."""
|
||||
from fastmcp.server.context import Context
|
||||
|
|
@ -289,4 +297,5 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
parameters=parameters,
|
||||
tags=tags or set(),
|
||||
enabled=enabled if enabled is not None else True,
|
||||
annotations=annotations,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from mcp.server.lowlevel.helper_types import ReadResourceContents
|
|||
from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import (
|
||||
Annotations,
|
||||
AnyFunction,
|
||||
CallToolRequestParams,
|
||||
ContentBlock,
|
||||
|
|
@ -1083,6 +1084,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
mime_type: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
annotations: Annotations | dict[str, Any] | None = None,
|
||||
) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
|
||||
"""Decorator to register a function as a resource.
|
||||
|
||||
|
|
@ -1106,6 +1108,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
mime_type: Optional MIME type for the resource
|
||||
tags: Optional set of tags for categorizing the resource
|
||||
enabled: Optional boolean to enable or disable the resource
|
||||
annotations: Optional annotations about the resource's behavior
|
||||
|
||||
Examples:
|
||||
Register a resource with a custom name:
|
||||
|
|
@ -1134,6 +1137,9 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
return f"Weather for {city}: {data}"
|
||||
```
|
||||
"""
|
||||
if isinstance(annotations, dict):
|
||||
annotations = Annotations(**annotations)
|
||||
|
||||
# Check if user passed function directly instead of calling decorator
|
||||
if inspect.isroutine(uri):
|
||||
raise TypeError(
|
||||
|
|
@ -1175,6 +1181,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
annotations=annotations,
|
||||
)
|
||||
self.add_template(template)
|
||||
return template
|
||||
|
|
@ -1188,6 +1195,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
annotations=annotations,
|
||||
)
|
||||
self.add_resource(resource)
|
||||
return resource
|
||||
|
|
|
|||
|
|
@ -1450,6 +1450,33 @@ class TestResource:
|
|||
result = await client.read_resource(AnyUrl("file://test.bin"))
|
||||
assert result[0].blob == base64.b64encode(b"Binary file data").decode() # type: ignore[attr-defined]
|
||||
|
||||
async def test_resource_with_annotations(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.resource(
|
||||
"http://example.com/data",
|
||||
name="test",
|
||||
annotations={
|
||||
"httpMethod": "GET",
|
||||
"Cache-Control": "max-age=3600",
|
||||
},
|
||||
)
|
||||
def get_data() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
resources = await client.list_resources()
|
||||
assert len(resources) == 1
|
||||
|
||||
resource = resources[0]
|
||||
assert str(resource.uri) == "http://example.com/data"
|
||||
|
||||
assert resource.annotations is not None
|
||||
assert hasattr(resource.annotations, "httpMethod")
|
||||
assert getattr(resource.annotations, "httpMethod") == "GET"
|
||||
assert hasattr(resource.annotations, "Cache-Control")
|
||||
assert getattr(resource.annotations, "Cache-Control") == "max-age=3600"
|
||||
|
||||
|
||||
class TestResourceTags:
|
||||
def create_server(self, include_tags=None, exclude_tags=None):
|
||||
|
|
@ -1848,6 +1875,30 @@ class TestResourceTemplates:
|
|||
result = await client.read_resource(AnyUrl("resource://a/b"))
|
||||
assert result[0].text == "Template resource 1: a/b" # type: ignore[attr-defined]
|
||||
|
||||
async def test_resource_template_with_annotations(self):
|
||||
"""Test that resource template annotations are visible to clients."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.resource(
|
||||
"api://users/{user_id}",
|
||||
annotations={"httpMethod": "GET", "Cache-Control": "no-cache"},
|
||||
)
|
||||
def get_user(user_id: str) -> str:
|
||||
return f"User {user_id} data"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
templates = await client.list_resource_templates()
|
||||
assert len(templates) == 1
|
||||
|
||||
template = templates[0]
|
||||
assert template.uriTemplate == "api://users/{user_id}"
|
||||
|
||||
assert template.annotations is not None
|
||||
assert hasattr(template.annotations, "httpMethod")
|
||||
assert getattr(template.annotations, "httpMethod") == "GET"
|
||||
assert hasattr(template.annotations, "Cache-Control")
|
||||
assert getattr(template.annotations, "Cache-Control") == "no-cache"
|
||||
|
||||
|
||||
class TestResourceTemplatesTags:
|
||||
def create_server(self, include_tags=None, exclude_tags=None):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue