feat: add tool metadata in FastMCP.tool decorator (#1463)
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>
This commit is contained in:
@@ -4,7 +4,14 @@ from __future__ import annotations as _annotations
|
||||
|
||||
import inspect
|
||||
import re
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable, Collection, Iterable, Sequence
|
||||
from collections.abc import (
|
||||
AsyncIterator,
|
||||
Awaitable,
|
||||
Callable,
|
||||
Collection,
|
||||
Iterable,
|
||||
Sequence,
|
||||
)
|
||||
from contextlib import AbstractAsyncContextManager, asynccontextmanager
|
||||
from typing import Any, Generic, Literal
|
||||
|
||||
@@ -22,10 +29,21 @@ from starlette.routing import Mount, Route
|
||||
from starlette.types import Receive, Scope, Send
|
||||
|
||||
from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
|
||||
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
|
||||
from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
|
||||
from mcp.server.auth.middleware.bearer_auth import (
|
||||
BearerAuthBackend,
|
||||
RequireAuthMiddleware,
|
||||
)
|
||||
from mcp.server.auth.provider import (
|
||||
OAuthAuthorizationServerProvider,
|
||||
ProviderTokenVerifier,
|
||||
TokenVerifier,
|
||||
)
|
||||
from mcp.server.auth.settings import AuthSettings
|
||||
from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, elicit_with_validation
|
||||
from mcp.server.elicitation import (
|
||||
ElicitationResult,
|
||||
ElicitSchemaModelT,
|
||||
elicit_with_validation,
|
||||
)
|
||||
from mcp.server.fastmcp.exceptions import ResourceError
|
||||
from mcp.server.fastmcp.prompts import Prompt, PromptManager
|
||||
from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager
|
||||
@@ -112,7 +130,9 @@ def lifespan_wrapper(
|
||||
lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]],
|
||||
) -> Callable[[MCPServer[LifespanResultT, Request]], AbstractAsyncContextManager[LifespanResultT]]:
|
||||
@asynccontextmanager
|
||||
async def wrap(_: MCPServer[LifespanResultT, Request]) -> AsyncIterator[LifespanResultT]:
|
||||
async def wrap(
|
||||
_: MCPServer[LifespanResultT, Request],
|
||||
) -> AsyncIterator[LifespanResultT]:
|
||||
async with lifespan(app) as context:
|
||||
yield context
|
||||
|
||||
@@ -126,7 +146,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
instructions: str | None = None,
|
||||
website_url: str | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
|
||||
auth_server_provider: (OAuthAuthorizationServerProvider[Any, Any, Any] | None) = None,
|
||||
token_verifier: TokenVerifier | None = None,
|
||||
event_store: EventStore | None = None,
|
||||
*,
|
||||
@@ -145,7 +165,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
warn_on_duplicate_tools: bool = True,
|
||||
warn_on_duplicate_prompts: bool = True,
|
||||
dependencies: Collection[str] = (),
|
||||
lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = None,
|
||||
lifespan: (Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None) = None,
|
||||
auth: AuthSettings | None = None,
|
||||
transport_security: TransportSecuritySettings | None = None,
|
||||
):
|
||||
@@ -290,6 +310,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
outputSchema=info.output_schema,
|
||||
annotations=info.annotations,
|
||||
icons=info.icons,
|
||||
_meta=info.meta,
|
||||
)
|
||||
for info in tools
|
||||
]
|
||||
@@ -365,6 +386,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
description: str | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
structured_output: bool | None = None,
|
||||
) -> None:
|
||||
"""Add a tool to the server.
|
||||
@@ -390,6 +412,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
description=description,
|
||||
annotations=annotations,
|
||||
icons=icons,
|
||||
meta=meta,
|
||||
structured_output=structured_output,
|
||||
)
|
||||
|
||||
@@ -411,6 +434,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
description: str | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
structured_output: bool | None = None,
|
||||
) -> Callable[[AnyFunction], AnyFunction]:
|
||||
"""Decorator to register a tool.
|
||||
@@ -458,6 +482,7 @@ class FastMCP(Generic[LifespanResultT]):
|
||||
description=description,
|
||||
annotations=annotations,
|
||||
icons=icons,
|
||||
meta=meta,
|
||||
structured_output=structured_output,
|
||||
)
|
||||
return fn
|
||||
@@ -1169,7 +1194,10 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]):
|
||||
"""
|
||||
|
||||
return await elicit_with_validation(
|
||||
session=self.request_context.session, message=message, schema=schema, related_request_id=self.request_id
|
||||
session=self.request_context.session,
|
||||
message=message,
|
||||
schema=schema,
|
||||
related_request_id=self.request_id,
|
||||
)
|
||||
|
||||
async def log(
|
||||
|
||||
@@ -34,6 +34,7 @@ class Tool(BaseModel):
|
||||
context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
|
||||
annotations: ToolAnnotations | None = Field(None, description="Optional annotations for the tool")
|
||||
icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this tool")
|
||||
meta: dict[str, Any] | None = Field(default=None, description="Optional metadata for this tool")
|
||||
|
||||
@cached_property
|
||||
def output_schema(self) -> dict[str, Any] | None:
|
||||
@@ -49,6 +50,7 @@ class Tool(BaseModel):
|
||||
context_kwarg: str | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
structured_output: bool | None = None,
|
||||
) -> Tool:
|
||||
"""Create a Tool from a function."""
|
||||
@@ -81,6 +83,7 @@ class Tool(BaseModel):
|
||||
context_kwarg=context_kwarg,
|
||||
annotations=annotations,
|
||||
icons=icons,
|
||||
meta=meta,
|
||||
)
|
||||
|
||||
async def run(
|
||||
|
||||
@@ -50,6 +50,7 @@ class ToolManager:
|
||||
description: str | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
icons: list[Icon] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
structured_output: bool | None = None,
|
||||
) -> Tool:
|
||||
"""Add a tool to the server."""
|
||||
@@ -60,6 +61,7 @@ class ToolManager:
|
||||
description=description,
|
||||
annotations=annotations,
|
||||
icons=icons,
|
||||
meta=meta,
|
||||
structured_output=structured_output,
|
||||
)
|
||||
existing = self._tools.get(tool.name)
|
||||
|
||||
@@ -635,6 +635,178 @@ class TestStructuredOutput:
|
||||
assert result == expected
|
||||
|
||||
|
||||
class TestToolMetadata:
|
||||
"""Test tool metadata functionality."""
|
||||
|
||||
def test_add_tool_with_metadata(self):
|
||||
"""Test adding a tool with metadata via ToolManager."""
|
||||
|
||||
def process_data(input_data: str) -> str:
|
||||
"""Process some data."""
|
||||
return f"Processed: {input_data}"
|
||||
|
||||
metadata = {"ui": {"type": "form", "fields": ["input"]}, "version": "1.0"}
|
||||
|
||||
manager = ToolManager()
|
||||
tool = manager.add_tool(process_data, meta=metadata)
|
||||
|
||||
assert tool.meta is not None
|
||||
assert tool.meta == metadata
|
||||
assert tool.meta["ui"]["type"] == "form"
|
||||
assert tool.meta["version"] == "1.0"
|
||||
|
||||
def test_add_tool_without_metadata(self):
|
||||
"""Test that tools without metadata have None as meta value."""
|
||||
|
||||
def simple_tool(x: int) -> int:
|
||||
"""Simple tool."""
|
||||
return x * 2
|
||||
|
||||
manager = ToolManager()
|
||||
tool = manager.add_tool(simple_tool)
|
||||
|
||||
assert tool.meta is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_in_fastmcp_decorator(self):
|
||||
"""Test that metadata is correctly added via FastMCP.tool decorator."""
|
||||
|
||||
app = FastMCP()
|
||||
|
||||
metadata = {"client": {"ui_component": "file_picker"}, "priority": "high"}
|
||||
|
||||
@app.tool(meta=metadata)
|
||||
def upload_file(filename: str) -> str:
|
||||
"""Upload a file."""
|
||||
return f"Uploaded: {filename}"
|
||||
|
||||
# Get the tool from the tool manager
|
||||
tool = app._tool_manager.get_tool("upload_file")
|
||||
assert tool is not None
|
||||
assert tool.meta is not None
|
||||
assert tool.meta == metadata
|
||||
assert tool.meta["client"]["ui_component"] == "file_picker"
|
||||
assert tool.meta["priority"] == "high"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_in_list_tools(self):
|
||||
"""Test that metadata is included in MCPTool when listing tools."""
|
||||
|
||||
app = FastMCP()
|
||||
|
||||
metadata = {
|
||||
"ui": {"input_type": "textarea", "rows": 5},
|
||||
"tags": ["text", "processing"],
|
||||
}
|
||||
|
||||
@app.tool(meta=metadata)
|
||||
def analyze_text(text: str) -> dict[str, Any]:
|
||||
"""Analyze text content."""
|
||||
return {"length": len(text), "words": len(text.split())}
|
||||
|
||||
tools = await app.list_tools()
|
||||
assert len(tools) == 1
|
||||
assert tools[0].meta is not None
|
||||
assert tools[0].meta == metadata
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multiple_tools_with_different_metadata(self):
|
||||
"""Test multiple tools with different metadata values."""
|
||||
|
||||
app = FastMCP()
|
||||
|
||||
metadata1 = {"ui": "form", "version": 1}
|
||||
metadata2 = {"ui": "picker", "experimental": True}
|
||||
|
||||
@app.tool(meta=metadata1)
|
||||
def tool1(x: int) -> int:
|
||||
"""First tool."""
|
||||
return x
|
||||
|
||||
@app.tool(meta=metadata2)
|
||||
def tool2(y: str) -> str:
|
||||
"""Second tool."""
|
||||
return y
|
||||
|
||||
@app.tool()
|
||||
def tool3(z: bool) -> bool:
|
||||
"""Third tool without metadata."""
|
||||
return z
|
||||
|
||||
tools = await app.list_tools()
|
||||
assert len(tools) == 3
|
||||
|
||||
# Find tools by name and check metadata
|
||||
tools_by_name = {t.name: t for t in tools}
|
||||
|
||||
assert tools_by_name["tool1"].meta == metadata1
|
||||
assert tools_by_name["tool2"].meta == metadata2
|
||||
assert tools_by_name["tool3"].meta is None
|
||||
|
||||
def test_metadata_with_complex_structure(self):
|
||||
"""Test metadata with complex nested structures."""
|
||||
|
||||
def complex_tool(data: str) -> str:
|
||||
"""Tool with complex metadata."""
|
||||
return data
|
||||
|
||||
metadata = {
|
||||
"ui": {
|
||||
"components": [
|
||||
{"type": "input", "name": "field1", "validation": {"required": True, "minLength": 5}},
|
||||
{"type": "select", "name": "field2", "options": ["a", "b", "c"]},
|
||||
],
|
||||
"layout": {"columns": 2, "responsive": True},
|
||||
},
|
||||
"permissions": ["read", "write"],
|
||||
"tags": ["data-processing", "user-input"],
|
||||
"version": 2,
|
||||
}
|
||||
|
||||
manager = ToolManager()
|
||||
tool = manager.add_tool(complex_tool, meta=metadata)
|
||||
|
||||
assert tool.meta is not None
|
||||
assert tool.meta["ui"]["components"][0]["validation"]["minLength"] == 5
|
||||
assert tool.meta["ui"]["layout"]["columns"] == 2
|
||||
assert "read" in tool.meta["permissions"]
|
||||
assert "data-processing" in tool.meta["tags"]
|
||||
|
||||
def test_metadata_empty_dict(self):
|
||||
"""Test that empty dict metadata is preserved."""
|
||||
|
||||
def tool_with_empty_meta(x: int) -> int:
|
||||
"""Tool with empty metadata."""
|
||||
return x
|
||||
|
||||
manager = ToolManager()
|
||||
tool = manager.add_tool(tool_with_empty_meta, meta={})
|
||||
|
||||
assert tool.meta is not None
|
||||
assert tool.meta == {}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_with_annotations(self):
|
||||
"""Test that metadata and annotations can coexist."""
|
||||
|
||||
app = FastMCP()
|
||||
|
||||
metadata = {"custom": "value"}
|
||||
annotations = ToolAnnotations(title="Combined Tool", readOnlyHint=True)
|
||||
|
||||
@app.tool(meta=metadata, annotations=annotations)
|
||||
def combined_tool(data: str) -> str:
|
||||
"""Tool with both metadata and annotations."""
|
||||
return data
|
||||
|
||||
tools = await app.list_tools()
|
||||
assert len(tools) == 1
|
||||
assert tools[0].meta == metadata
|
||||
assert tools[0].annotations is not None
|
||||
assert tools[0].annotations.title == "Combined Tool"
|
||||
assert tools[0].annotations.readOnlyHint is True
|
||||
|
||||
|
||||
class TestRemoveTools:
|
||||
"""Test tool removal functionality in the tool manager."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user