diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 485ef1519..4ed052efd 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -4,7 +4,14 @@ 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.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 @@ def __init__( # noqa: PLR0913 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 @@ def __init__( # noqa: PLR0913 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 @@ async def list_tools(self) -> list[MCPTool]: outputSchema=info.output_schema, annotations=info.annotations, icons=info.icons, + _meta=info.meta, ) for info in tools ] @@ -363,6 +384,7 @@ def add_tool( 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. @@ -388,6 +410,7 @@ def add_tool( description=description, annotations=annotations, icons=icons, + meta=meta, structured_output=structured_output, ) @@ -409,6 +432,7 @@ def tool( 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. @@ -456,6 +480,7 @@ def decorator(fn: AnyFunction) -> AnyFunction: description=description, annotations=annotations, icons=icons, + meta=meta, structured_output=structured_output, ) return fn @@ -1164,7 +1189,10 @@ async def elicit( """ 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( diff --git a/src/mcp/server/fastmcp/tools/base.py b/src/mcp/server/fastmcp/tools/base.py index 3f26ddcea..7002af493 100644 --- a/src/mcp/server/fastmcp/tools/base.py +++ b/src/mcp/server/fastmcp/tools/base.py @@ -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 @@ def from_function( 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 @@ def from_function( context_kwarg=context_kwarg, annotations=annotations, icons=icons, + meta=meta, ) async def run( diff --git a/src/mcp/server/fastmcp/tools/tool_manager.py b/src/mcp/server/fastmcp/tools/tool_manager.py index d6c0054af..095753de6 100644 --- a/src/mcp/server/fastmcp/tools/tool_manager.py +++ b/src/mcp/server/fastmcp/tools/tool_manager.py @@ -50,6 +50,7 @@ def add_tool( 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 @@ def add_tool( description=description, annotations=annotations, icons=icons, + meta=meta, structured_output=structured_output, ) existing = self._tools.get(tool.name) diff --git a/tests/server/fastmcp/test_tool_manager.py b/tests/server/fastmcp/test_tool_manager.py index 71884fba2..7a53ec37a 100644 --- a/tests/server/fastmcp/test_tool_manager.py +++ b/tests/server/fastmcp/test_tool_manager.py @@ -635,6 +635,178 @@ def get_scores() -> dict[str, int]: 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."""