diff --git a/src/fastmcp/prompts/prompt.py b/src/fastmcp/prompts/prompt.py index 06cc23301..5733b3705 100644 --- a/src/fastmcp/prompts/prompt.py +++ b/src/fastmcp/prompts/prompt.py @@ -7,6 +7,8 @@ from typing import Annotated, Any, Literal import pydantic_core from mcp.types import EmbeddedResource, ImageContent, TextContent +from mcp.types import Prompt as MCPPrompt +from mcp.types import PromptArgument as MCPPromptArgument from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call from fastmcp.utilities.types import _convert_set_defaults @@ -166,3 +168,20 @@ class Prompt(BaseModel): if not isinstance(other, Prompt): return False return self.model_dump() == other.model_dump() + + def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt: + """Convert the prompt to an MCP prompt.""" + arguments = [ + MCPPromptArgument( + name=arg.name, + description=arg.description, + required=arg.required, + ) + for arg in self.arguments or [] + ] + kwargs = { + "name": self.name, + "description": self.description, + "arguments": arguments, + } + return MCPPrompt(**kwargs | overrides) diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py index 30f971589..ce423c731 100644 --- a/src/fastmcp/prompts/prompt_manager.py +++ b/src/fastmcp/prompts/prompt_manager.py @@ -1,11 +1,10 @@ """Prompt management functionality.""" -import copy from collections.abc import Awaitable, Callable from typing import Any from fastmcp.exceptions import PromptError -from fastmcp.prompts.prompt import Message, Prompt, PromptResult +from fastmcp.prompts.prompt import MCPPrompt, Message, Prompt, PromptResult from fastmcp.settings import DuplicateBehavior from fastmcp.utilities.logging import get_logger @@ -30,13 +29,23 @@ class PromptManager: self.duplicate_behavior = duplicate_behavior - def get_prompt(self, name: str) -> Prompt | None: - """Get prompt by name.""" - return self._prompts.get(name) + def get_prompt(self, key: str) -> Prompt | None: + """Get prompt by key.""" + return self._prompts.get(key) + + def get_prompts(self) -> dict[str, Prompt]: + """Get all registered prompts, indexed by registered key.""" + return self._prompts def list_prompts(self) -> list[Prompt]: """List all registered prompts.""" - return list(self._prompts.values()) + return list(self.get_prompts().values()) + + def list_mcp_prompts(self) -> list[MCPPrompt]: + """List all registered prompts in the format expected by the low-level MCP server.""" + return [ + prompt.to_mcp_prompt(name=key) for key, prompt in self.get_prompts().items() + ] def add_prompt_from_fn( self, @@ -49,23 +58,24 @@ class PromptManager: prompt = Prompt.from_function(fn, name=name, description=description, tags=tags) return self.add_prompt(prompt) - def add_prompt(self, prompt: Prompt) -> Prompt: + def add_prompt(self, prompt: Prompt, key: str | None = None) -> Prompt: """Add a prompt to the manager.""" + key = key or prompt.name # Check for duplicates - existing = self._prompts.get(prompt.name) + existing = self._prompts.get(key) if existing: if self.duplicate_behavior == "warn": - logger.warning(f"Prompt already exists: {prompt.name}") - self._prompts[prompt.name] = prompt + logger.warning(f"Prompt already exists: {key}") + self._prompts[key] = prompt elif self.duplicate_behavior == "replace": - self._prompts[prompt.name] = prompt + self._prompts[key] = prompt elif self.duplicate_behavior == "error": - raise ValueError(f"Prompt already exists: {prompt.name}") + raise ValueError(f"Prompt already exists: {key}") elif self.duplicate_behavior == "ignore": return existing else: - self._prompts[prompt.name] = prompt + self._prompts[key] = prompt return prompt async def render_prompt( @@ -86,19 +96,16 @@ class PromptManager: Args: manager: Another PromptManager instance to import prompts from - prefix: Prefix to add to prompt names. The resulting prompt name will + prefix: Prefix to add to prompt names. The resulting prompt key will be in the format "{prefix}{original_name}" if prefix is provided, otherwise the original name is used. For example, with prefix "weather/" and prompt "forecast_prompt", the imported prompt would be available as "weather/forecast_prompt" """ for name, prompt in manager._prompts.items(): - # Create prefixed name - prefixed_name = f"{prefix}{name}" if prefix else name + # Create prefixed key + key = f"{prefix}{name}" if prefix else name - new_prompt = copy.copy(prompt) - new_prompt.name = prefixed_name - - # Store the prompt with the prefixed name - self.add_prompt(new_prompt) - logger.debug(f'Imported prompt "{name}" as "{prefixed_name}"') + # Store the prompt with the prefixed key + self.add_prompt(prompt, key=key) + logger.debug(f'Imported prompt "{prompt.name}" as "{key}"') diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index b99e5d867..d1c04087d 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -28,7 +28,6 @@ from mcp.types import ( TextContent, ) from mcp.types import Prompt as MCPPrompt -from mcp.types import PromptArgument as MCPPromptArgument from mcp.types import Resource as MCPResource from mcp.types import ResourceTemplate as MCPResourceTemplate from mcp.types import Tool as MCPTool @@ -217,11 +216,11 @@ class FastMCP(Generic[LifespanResultT]): return Context(request_context=request_context, fastmcp=self) async def call_tool( - self, name: str, arguments: dict[str, Any] + self, key: str, arguments: dict[str, Any] ) -> list[TextContent | ImageContent | EmbeddedResource]: """Call a tool by name with arguments.""" context = self.get_context() - result = await self._tool_manager.call_tool(name, arguments, context=context) + result = await self._tool_manager.call_tool(key, arguments, context=context) converted_result = _convert_to_content(result) return converted_result @@ -592,22 +591,7 @@ class FastMCP(Generic[LifespanResultT]): See `list_prompts` for a more ergonomic way to list prompts. """ - prompts = self.list_prompts() - return [ - MCPPrompt( - name=prompt.name, - description=prompt.description, - arguments=[ - MCPPromptArgument( - name=arg.name, - description=arg.description, - required=arg.required, - ) - for arg in (prompt.arguments or []) - ], - ) - for prompt in prompts - ] + return self._prompt_manager.list_mcp_prompts() async def get_prompt( self, name: str, arguments: dict[str, Any] | None = None diff --git a/src/fastmcp/tools/tool_manager.py b/src/fastmcp/tools/tool_manager.py index 9e0f1ac42..f11d98e22 100644 --- a/src/fastmcp/tools/tool_manager.py +++ b/src/fastmcp/tools/tool_manager.py @@ -36,9 +36,9 @@ class ToolManager: self.duplicate_behavior = duplicate_behavior - def get_tool(self, name: str) -> Tool | None: - """Get tool by name.""" - return self._tools.get(name) + def get_tool(self, key: str) -> Tool | None: + """Get tool by key.""" + return self._tools.get(key) def get_tools(self) -> dict[str, Tool]: """Get all registered tools, indexed by registered key.""" @@ -109,6 +109,6 @@ class ToolManager: the imported tool would be available as "weather/forecast" """ for name, tool in tool_manager._tools.items(): - prefixed_name = f"{prefix}{name}" if prefix else name - self.add_tool(tool, key=prefixed_name) - logger.debug(f'Imported tool "{tool.name}" as "{prefixed_name}"') + key = f"{prefix}{name}" if prefix else name + self.add_tool(tool, key=key) + logger.debug(f'Imported tool "{tool.name}" as "{key}"')