Add support for prompts

examples/echo.py

@@ -10,10 +10,21 @@ mcp = FastMCP("Echo Server")
 
 
 @mcp.tool()
-def echo(text: str) -> str:
+def echo_tool(text: str) -> str:
     """Echo the input text"""
     return text
 
 
+@mcp.resource("echo://{text}")
+def echo_resource(text: str) -> str:
+    """Echo the input text"""
+    return f"Echo: {text}"
+
+
+@mcp.prompt("echo")
+def echo_prompt(text: str) -> str:
+    return text
+
+
 if __name__ == "__main__":
     mcp.run()

examples/simple_echo.py

@@ -0,0 +1,19 @@
+"""
+FastMCP Echo Server
+"""
+
+from fastmcp import FastMCP
+
+
+# Create server
+mcp = FastMCP("Echo Server")
+
+
+@mcp.tool()
+def echo(text: str) -> str:
+    """Echo the input text"""
+    return text
+
+
+if __name__ == "__main__":
+    mcp.run()

pyproject.toml

@@ -5,9 +5,9 @@ description = "A more ergonomic interface for MCP servers"
 authors = [{ name = "Jeremiah Lowin" }]
 dependencies = [
     "httpx>=0.26.0",
-    "mcp>=1.0.0",
+    "mcp>=1.0.0,<2.0.0",
     "pydantic-settings>=2.6.1",
-    "pydantic>=2.5.3",
+    "pydantic>=2.5.3,<3.0.0",
     "typer>=0.9.0",
 ]
 requires-python = ">=3.10"

src/fastmcp/prompts/__init__.py

@@ -0,0 +1,4 @@
+from .base import Prompt
+from .manager import PromptManager
+
+__all__ = ["Prompt", "PromptManager"]

src/fastmcp/prompts/base.py

@@ -0,0 +1,149 @@
+"""Base classes for FastMCP prompts."""
+
+import json
+from typing import Any, Callable, Dict, Literal, Optional, Sequence, Union
+import inspect
+
+from pydantic import BaseModel, Field, TypeAdapter, field_validator, validate_call
+from mcp.types import TextContent, ImageContent, EmbeddedResource
+import pydantic_core
+
+
+class Message(BaseModel):
+    """Base class for all prompt messages."""
+
+    role: Literal["user", "assistant"]
+    content: Union[TextContent, ImageContent, EmbeddedResource]
+
+    def __init__(self, content, **kwargs):
+        super().__init__(content=content, **kwargs)
+
+    @field_validator("content", mode="before")
+    def validate_content(cls, v):
+        if isinstance(v, str):
+            return TextContent(type="text", text=v)
+        return v
+
+
+class UserMessage(Message):
+    """A message from the user."""
+
+    role: Literal["user"] = "user"
+
+
+class AssistantMessage(Message):
+    """A message from the assistant."""
+
+    role: Literal["assistant"] = "assistant"
+
+
+message_validator = TypeAdapter(Union[UserMessage, AssistantMessage])
+
+
+class PromptArgument(BaseModel):
+    """An argument that can be passed to a prompt."""
+
+    name: str = Field(description="Name of the argument")
+    description: str | None = Field(
+        None, description="Description of what the argument does"
+    )
+    required: bool = Field(
+        default=False, description="Whether the argument is required"
+    )
+
+
+class Prompt(BaseModel):
+    """A prompt template that can be rendered with parameters."""
+
+    name: str = Field(description="Name of the prompt")
+    description: str | None = Field(
+        None, description="Description of what the prompt does"
+    )
+    arguments: list[PromptArgument] | None = Field(
+        None, description="Arguments that can be passed to the prompt"
+    )
+    fn: Callable = Field(exclude=True)
+
+    @classmethod
+    def from_function(
+        cls,
+        fn: Callable[..., Sequence[Message]],
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> "Prompt":
+        """Create a Prompt from a function."""
+        func_name = name or fn.__name__
+
+        if func_name == "<lambda>":
+            raise ValueError("You must provide a name for lambda functions")
+
+        # Get schema from TypeAdapter - will fail if function isn't properly typed
+        parameters = TypeAdapter(fn).json_schema()
+
+        # Convert parameters to PromptArguments
+        arguments = []
+        if "properties" in parameters:
+            for param_name, param in parameters["properties"].items():
+                required = param_name in parameters.get("required", [])
+                arguments.append(
+                    PromptArgument(
+                        name=param_name,
+                        description=param.get("description"),
+                        required=required,
+                    )
+                )
+
+        # ensure the arguments are properly cast
+        fn = validate_call(fn)
+
+        return cls(
+            name=func_name,
+            description=description or fn.__doc__ or "",
+            arguments=arguments,
+            fn=fn,
+        )
+
+    async def render(self, arguments: Optional[Dict[str, Any]] = None) -> list[Message]:
+        """Render the prompt with arguments."""
+        # Validate required arguments
+        if self.arguments:
+            required = {arg.name for arg in self.arguments if arg.required}
+            provided = set(arguments or {})
+            missing = required - provided
+            if missing:
+                raise ValueError(f"Missing required arguments: {missing}")
+
+        try:
+            # Call function and check if result is a coroutine
+            result = self.fn(**(arguments or {}))
+            if inspect.iscoroutine(result):
+                result = await result
+
+            # Validate messages
+            if not isinstance(result, (list, tuple)):
+                result = [result]
+
+            # Convert result to messages
+            messages = []
+            for msg in result:
+                try:
+                    if isinstance(msg, Message):
+                        messages.append(msg)
+                    elif isinstance(msg, dict):
+                        msg = message_validator.validate_python(msg)
+                        messages.append(msg)
+                    elif isinstance(msg, str):
+                        messages.append(
+                            UserMessage(content=TextContent(type="text", text=msg))
+                        )
+                    else:
+                        msg = json.dumps(pydantic_core.to_jsonable_python(msg))
+                        messages.append(Message(role="user", content=msg))
+                except Exception:
+                    raise ValueError(
+                        f"Could not convert prompt result to message: {msg}"
+                    )
+
+            return messages
+        except Exception as e:
+            raise ValueError(f"Error rendering prompt {self.name}: {e}")

src/fastmcp/prompts/manager.py

@@ -0,0 +1,50 @@
+"""Prompt management functionality."""
+
+from typing import Any, Dict, Optional
+
+from fastmcp.prompts.base import Message, Prompt
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class PromptManager:
+    """Manages FastMCP prompts."""
+
+    def __init__(self, warn_on_duplicate_prompts: bool = True):
+        self._prompts: Dict[str, Prompt] = {}
+        self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
+
+    def get_prompt(self, name: str) -> Optional[Prompt]:
+        """Get prompt by name."""
+        return self._prompts.get(name)
+
+    def list_prompts(self) -> list[Prompt]:
+        """List all registered prompts."""
+        return list(self._prompts.values())
+
+    def add_prompt(
+        self,
+        prompt: Prompt,
+    ) -> Prompt:
+        """Add a prompt to the manager."""
+
+        # Check for duplicates
+        existing = self._prompts.get(prompt.name)
+        if existing:
+            if self.warn_on_duplicate_prompts:
+                logger.warning(f"Prompt already exists: {prompt.name}")
+            return existing
+
+        self._prompts[prompt.name] = prompt
+        return prompt
+
+    async def render_prompt(
+        self, name: str, arguments: Optional[Dict[str, Any]] = None
+    ) -> list[Message]:
+        """Render a prompt by name with arguments."""
+        prompt = self.get_prompt(name)
+        if not prompt:
+            raise ValueError(f"Unknown prompt: {name}")
+
+        return await prompt.render(arguments)

src/fastmcp/prompts/prompt_manager.py

@@ -0,0 +1,36 @@
+"""Prompt management functionality."""
+
+from typing import Dict, Optional
+
+
+from fastmcp.prompts.base import Prompt
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class PromptManager:
+    """Manages FastMCP prompts."""
+
+    def __init__(self, warn_on_duplicate_prompts: bool = True):
+        self._prompts: Dict[str, Prompt] = {}
+        self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
+
+    def add_prompt(self, prompt: Prompt) -> Prompt:
+        """Add a prompt to the manager."""
+        logger.debug(f"Adding prompt: {prompt.name}")
+        existing = self._prompts.get(prompt.name)
+        if existing:
+            if self.warn_on_duplicate_prompts:
+                logger.warning(f"Prompt already exists: {prompt.name}")
+            return existing
+        self._prompts[prompt.name] = prompt
+        return prompt
+
+    def get_prompt(self, name: str) -> Optional[Prompt]:
+        """Get prompt by name."""
+        return self._prompts.get(name)
+
+    def list_prompts(self) -> list[Prompt]:
+        """List all registered prompts."""
+        return list(self._prompts.values())

src/fastmcp/server.py

@@ -1,7 +1,7 @@
 """FastMCP - A more ergonomic interface for MCP servers."""
 
 import pydantic_core
-from typing import Any, Literal, Optional, Union
+from typing import Any, Literal, Optional, Union, Dict
 
 from mcp.server import RequestContext
 from pydantic import BaseModel
@@ -24,6 +24,9 @@ from mcp.types import (
     ResourceTemplate as MCPResourceTemplate,
     TextContent,
     ImageContent,
+    EmbeddedResource,
+    Prompt as MCPPrompt,
+    GetPromptResult,
 )
 from pydantic_settings import BaseSettings
 from pydantic.networks import _BaseUrl
@@ -33,6 +36,8 @@ from fastmcp.resources import Resource, ResourceManager, FunctionResource
 from fastmcp.tools import ToolManager
 from fastmcp.utilities.logging import configure_logging
 from fastmcp.utilities.types import Image
+from fastmcp.prompts import Prompt, PromptManager
+from itertools import chain
 
 logger = get_logger(__name__)
 
@@ -60,6 +65,9 @@ class Settings(BaseSettings):
     # tool settings
     warn_on_duplicate_tools: bool = True
 
+    # prompt settings
+    warn_on_duplicate_prompts: bool = True
+
 
 class FastMCP:
     def __init__(self, name=None, **settings: Optional[Settings]):
@@ -71,6 +79,9 @@ class FastMCP:
         self._resource_manager = ResourceManager(
             warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
         )
+        self._prompt_manager = PromptManager(
+            warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts
+        )
 
         # Set up MCP protocol handlers
         self._setup_handlers()
@@ -103,6 +114,8 @@ class FastMCP:
         self._mcp_server.call_tool()(self.call_tool)
         self._mcp_server.list_resources()(self.list_resources)
         self._mcp_server.read_resource()(self.read_resource)
+        self._mcp_server.list_prompts()(self.list_prompts)
+        self._mcp_server.get_prompt()(self.get_prompt)
         # TODO: This has not been added to MCP yet, see https://github.com/jlowin/fastmcp/issues/10
         # self._mcp_server.list_resource_templates()(self.list_resource_templates)
 
@@ -133,21 +146,10 @@ class FastMCP:
         self, name: str, arguments: dict
     ) -> Sequence[Union[TextContent, ImageContent]]:
         """Call a tool by name with arguments."""
-        try:
-            context = self.get_context()
-            result = await self._tool_manager.call_tool(
-                name, arguments, context=context
-            )
-            return _convert_to_content(result)
-        except Exception as e:
-            logger.error(f"Error calling tool {name}: {e}")
-            return [
-                TextContent(
-                    type="text",
-                    text=str(e),
-                    is_error=True,
-                )
-            ]
+        context = self.get_context()
+        result = await self._tool_manager.call_tool(name, arguments, context=context)
+        converted_result = _convert_to_content(result)
+        return converted_result
 
     async def list_resources(self) -> list[MCPResource]:
         """List all available resources."""
@@ -335,6 +337,64 @@ class FastMCP:
 
         return decorator
 
+    def add_prompt(self, prompt: Prompt) -> None:
+        """Add a prompt to the server.
+
+        Args:
+            prompt: A Prompt instance to add
+        """
+        self._prompt_manager.add_prompt(prompt)
+
+    def prompt(
+        self, name: Optional[str] = None, description: Optional[str] = None
+    ) -> Callable:
+        """Decorator to register a prompt.
+
+        Args:
+            name: Optional name for the prompt (defaults to function name)
+            description: Optional description of what the prompt does
+
+        Example:
+            @server.prompt()
+            def analyze_table(table_name: str) -> list[Message]:
+                schema = read_table_schema(table_name)
+                return [
+                    {
+                        "role": "user",
+                        "content": f"Analyze this schema:\n{schema}"
+                    }
+                ]
+
+            @server.prompt()
+            async def analyze_file(path: str) -> list[Message]:
+                content = await read_file(path)
+                return [
+                    {
+                        "role": "user",
+                        "content": {
+                            "type": "resource",
+                            "resource": {
+                                "uri": f"file://{path}",
+                                "text": content
+                            }
+                        }
+                    }
+                ]
+        """
+        # Check if user passed function directly instead of calling decorator
+        if callable(name):
+            raise TypeError(
+                "The @prompt decorator was used incorrectly. "
+                "Did you forget to call it? Use @prompt() instead of @prompt"
+            )
+
+        def decorator(func: Callable) -> Callable:
+            prompt = Prompt.from_function(func, name=name, description=description)
+            self.add_prompt(prompt)
+            return func
+
+        return decorator
+
     async def run_stdio_async(self) -> None:
         """Run the server using stdio transport."""
         async with stdio_server() as (read_stream, write_stream):
@@ -381,45 +441,61 @@ class FastMCP:
             log_level=self.settings.log_level,
         )
 
+    async def list_prompts(self) -> list[MCPPrompt]:
+        """List all available prompts."""
+        prompts = self._prompt_manager.list_prompts()
+        return [
+            MCPPrompt(
+                name=prompt.name,
+                description=prompt.description,
+                arguments=[
+                    {
+                        "name": arg.name,
+                        "description": arg.description,
+                        "required": arg.required,
+                    }
+                    for arg in (prompt.arguments or [])
+                ],
+            )
+            for prompt in prompts
+        ]
 
-def _convert_to_content(value: Any) -> Sequence[Union[TextContent, ImageContent]]:
-    """Convert a tool result to MCP content types."""
-
-    # Already a sequence of valid content types
-    if isinstance(value, (list, tuple)):
-        if all(isinstance(x, (TextContent, ImageContent)) for x in value):
-            return value
-        # Handle mixed content including Image objects
-        result = []
-        for item in value:
-            if isinstance(item, (TextContent, ImageContent)):
-                result.append(item)
-            elif isinstance(item, Image):
-                result.append(item.to_image_content())
-            else:
-                result.append(
-                    TextContent(
-                        type="text",
-                        text=json.dumps(pydantic_core.to_jsonable_python(item)),
-                    )
-                )
-        return result
+    async def get_prompt(
+        self, name: str, arguments: Optional[Dict[str, Any]] = None
+    ) -> GetPromptResult:
+        """Get a prompt by name with arguments."""
+        try:
+            messages = await self._prompt_manager.render_prompt(name, arguments)
 
-    # Single content type
-    if isinstance(value, (TextContent, ImageContent)):
-        return [value]
+            return GetPromptResult(messages=pydantic_core.to_jsonable_python(messages))
+        except Exception as e:
+            logger.error(f"Error getting prompt {name}: {e}")
+            raise ValueError(str(e))
 
-    # Image helper
-    if isinstance(value, Image):
-        return [value.to_image_content()]
 
-    # All other types - convert to JSON string with pydantic encoder
-    return [
-        TextContent(
-            type="text",
-            text=json.dumps(pydantic_core.to_jsonable_python(value)),
-        )
-    ]
+def _convert_to_content(
+    result: Any,
+) -> Sequence[Union[TextContent, ImageContent, EmbeddedResource]]:
+    """Convert a result to a sequence of content objects."""
+    if result is None:
+        return []
+
+    if isinstance(result, (TextContent, ImageContent, EmbeddedResource)):
+        return [result]
+
+    if isinstance(result, Image):
+        return [result.to_image_content()]
+
+    if isinstance(result, (list, tuple)):
+        return list(chain.from_iterable(_convert_to_content(item) for item in result))
+
+    if not isinstance(result, str):
+        try:
+            result = json.dumps(pydantic_core.to_jsonable_python(result))
+        except Exception:
+            result = str(result)
+
+    return [TextContent(type="text", text=result)]
 
 
 class Context(BaseModel):

tests/prompts/test_base.py

@@ -0,0 +1,193 @@
+import pytest
+from fastmcp.prompts.base import (
+    Prompt,
+    UserMessage,
+    TextContent,
+    AssistantMessage,
+    Message,
+)
+from mcp.types import EmbeddedResource, TextResourceContents
+
+
+class TestRenderPrompt:
+    async def test_basic_fn(self):
+        def fn() -> str:
+            return "Hello, world!"
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    async def test_async_fn(self):
+        async def fn() -> str:
+            return "Hello, world!"
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    async def test_fn_with_args(self):
+        async def fn(name: str, age: int = 30) -> str:
+            return f"Hello, {name}! You're {age} years old."
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render(arguments=dict(name="World")) == [
+            UserMessage(
+                content=TextContent(
+                    type="text", text="Hello, World! You're 30 years old."
+                )
+            )
+        ]
+
+    async def test_fn_with_invalid_kwargs(self):
+        async def fn(name: str, age: int = 30) -> str:
+            return f"Hello, {name}! You're {age} years old."
+
+        prompt = Prompt.from_function(fn)
+        with pytest.raises(ValueError):
+            await prompt.render(arguments=dict(age=40))
+
+    async def test_fn_returns_message(self):
+        async def fn() -> UserMessage:
+            return UserMessage(content="Hello, world!")
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    async def test_fn_returns_assistant_message(self):
+        async def fn() -> AssistantMessage:
+            return AssistantMessage(
+                content=TextContent(type="text", text="Hello, world!")
+            )
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            AssistantMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    async def test_fn_returns_multiple_messages(self):
+        expected = [
+            UserMessage("Hello, world!"),
+            AssistantMessage("How can I help you today?"),
+            UserMessage("I'm looking for a restaurant in the center of town."),
+        ]
+
+        async def fn() -> list[Message]:
+            return expected
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == expected
+
+    async def test_fn_returns_list_of_strings(self):
+        expected = [
+            "Hello, world!",
+            "I'm looking for a restaurant in the center of town.",
+        ]
+
+        async def fn() -> list[str]:
+            return expected
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [UserMessage(t) for t in expected]
+
+    async def test_fn_returns_resource_content(self):
+        """Test returning a message with resource content."""
+
+        async def fn() -> UserMessage:
+            return UserMessage(
+                content=EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="file://file.txt",
+                        text="File contents",
+                        mimeType="text/plain",
+                    ),
+                )
+            )
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(
+                content=EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="file://file.txt",
+                        text="File contents",
+                        mimeType="text/plain",
+                    ),
+                )
+            )
+        ]
+
+    async def test_fn_returns_mixed_content(self):
+        """Test returning messages with mixed content types."""
+
+        async def fn() -> list[Message]:
+            return [
+                UserMessage(content="Please analyze this file:"),
+                UserMessage(
+                    content=EmbeddedResource(
+                        type="resource",
+                        resource=TextResourceContents(
+                            uri="file://file.txt",
+                            text="File contents",
+                            mimeType="text/plain",
+                        ),
+                    )
+                ),
+                AssistantMessage(content="I'll help analyze that file."),
+            ]
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(
+                content=TextContent(type="text", text="Please analyze this file:")
+            ),
+            UserMessage(
+                content=EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="file://file.txt",
+                        text="File contents",
+                        mimeType="text/plain",
+                    ),
+                )
+            ),
+            AssistantMessage(
+                content=TextContent(type="text", text="I'll help analyze that file.")
+            ),
+        ]
+
+    async def test_fn_returns_dict_with_resource(self):
+        """Test returning a dict with resource content."""
+
+        async def fn() -> dict:
+            return {
+                "role": "user",
+                "content": {
+                    "type": "resource",
+                    "resource": {
+                        "uri": "file://file.txt",
+                        "text": "File contents",
+                        "mimeType": "text/plain",
+                    },
+                },
+            }
+
+        prompt = Prompt.from_function(fn)
+        assert await prompt.render() == [
+            UserMessage(
+                content=EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="file://file.txt",
+                        text="File contents",
+                        mimeType="text/plain",
+                    ),
+                )
+            )
+        ]

tests/prompts/test_manager.py

@@ -0,0 +1,107 @@
+import pytest
+from fastmcp.prompts.base import UserMessage, TextContent, Prompt
+from fastmcp.prompts.manager import PromptManager
+
+
+class TestPromptManager:
+    def test_add_prompt(self):
+        """Test adding a prompt to the manager."""
+
+        def fn() -> str:
+            return "Hello, world!"
+
+        manager = PromptManager()
+        prompt = Prompt.from_function(fn)
+        added = manager.add_prompt(prompt)
+        assert added == prompt
+        assert manager.get_prompt("fn") == prompt
+
+    def test_add_duplicate_prompt(self, caplog):
+        """Test adding the same prompt twice."""
+
+        def fn() -> str:
+            return "Hello, world!"
+
+        manager = PromptManager()
+        prompt = Prompt.from_function(fn)
+        first = manager.add_prompt(prompt)
+        second = manager.add_prompt(prompt)
+        assert first == second
+        assert "Prompt already exists" in caplog.text
+
+    def test_disable_warn_on_duplicate_prompts(self, caplog):
+        """Test disabling warning on duplicate prompts."""
+
+        def fn() -> str:
+            return "Hello, world!"
+
+        manager = PromptManager(warn_on_duplicate_prompts=False)
+        prompt = Prompt.from_function(fn)
+        first = manager.add_prompt(prompt)
+        second = manager.add_prompt(prompt)
+        assert first == second
+        assert "Prompt already exists" not in caplog.text
+
+    def test_list_prompts(self):
+        """Test listing all prompts."""
+
+        def fn1() -> str:
+            return "Hello, world!"
+
+        def fn2() -> str:
+            return "Goodbye, world!"
+
+        manager = PromptManager()
+        prompt1 = Prompt.from_function(fn1)
+        prompt2 = Prompt.from_function(fn2)
+        manager.add_prompt(prompt1)
+        manager.add_prompt(prompt2)
+        prompts = manager.list_prompts()
+        assert len(prompts) == 2
+        assert prompts == [prompt1, prompt2]
+
+    async def test_render_prompt(self):
+        """Test rendering a prompt."""
+
+        def fn() -> str:
+            return "Hello, world!"
+
+        manager = PromptManager()
+        prompt = Prompt.from_function(fn)
+        manager.add_prompt(prompt)
+        messages = await manager.render_prompt("fn")
+        assert messages == [
+            UserMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    async def test_render_prompt_with_args(self):
+        """Test rendering a prompt with arguments."""
+
+        def fn(name: str) -> str:
+            return f"Hello, {name}!"
+
+        manager = PromptManager()
+        prompt = Prompt.from_function(fn)
+        manager.add_prompt(prompt)
+        messages = await manager.render_prompt("fn", arguments={"name": "World"})
+        assert messages == [
+            UserMessage(content=TextContent(type="text", text="Hello, World!"))
+        ]
+
+    async def test_render_unknown_prompt(self):
+        """Test rendering a non-existent prompt."""
+        manager = PromptManager()
+        with pytest.raises(ValueError, match="Unknown prompt: unknown"):
+            await manager.render_prompt("unknown")
+
+    async def test_render_prompt_with_missing_args(self):
+        """Test rendering a prompt with missing required arguments."""
+
+        def fn(name: str) -> str:
+            return f"Hello, {name}!"
+
+        manager = PromptManager()
+        prompt = Prompt.from_function(fn)
+        manager.add_prompt(prompt)
+        with pytest.raises(ValueError, match="Missing required arguments"):
+            await manager.render_prompt("fn")

tests/test_server.py

@@ -1,12 +1,13 @@
 from mcp.shared.memory import (
     create_connected_server_and_client_session as client_session,
 )
+from mcp.shared.exceptions import McpError
 from fastmcp import FastMCP, Context
 from fastmcp.resources import FileResource, FunctionResource
 from fastmcp.utilities.types import Image
 from mcp.types import TextContent, ImageContent
+from fastmcp.prompts.base import Message, UserMessage, TextContent, EmbeddedResource
 import pytest
-from pydantic import BaseModel
 from pathlib import Path
 import base64
 from typing import Union, TYPE_CHECKING
@@ -67,11 +68,6 @@ def error_tool_fn() -> None:
     raise ValueError("Test error")
 
 
-class ErrorResponse(BaseModel):
-    is_error: bool = True
-    message: str
-
-
 def image_tool_fn(path: str) -> Image:
     return Image(path)
 
@@ -113,7 +109,7 @@ class TestServerTools:
             assert len(result.content) == 1
             assert result.content[0].type == "text"
             assert "Test error" in result.content[0].text
-            assert result.content[0].is_error is True
+            assert result.isError is True
 
     async def test_tool_exception_content(self):
         """Test that exception details are properly formatted in the response"""
@@ -121,11 +117,10 @@ class TestServerTools:
         mcp.add_tool(error_tool_fn)
         async with client_session(mcp._mcp_server) as client:
             result = await client.call_tool("error_tool_fn", {})
-            content = result.content[0]
-            assert content.type == "text"
-            assert isinstance(content.text, str)
-            assert "Test error" in content.text
-            assert content.is_error is True
+            assert result.content[0].type == "text"
+            assert isinstance(result.content[0].text, str)
+            assert "Test error" in result.content[0].text
+            assert result.isError is True
 
     async def test_tool_text_conversion(self):
         mcp = FastMCP()
@@ -185,7 +180,7 @@ class TestServerTools:
             assert len(result.content) == 4
             # Check text conversion
             assert result.content[0].type == "text"
-            assert '"text message"' in result.content[0].text
+            assert "text message" in result.content[0].text
             # Check image conversion
             assert result.content[1].type == "image"
             assert result.content[1].mimeType == "image/png"
@@ -463,3 +458,134 @@ class TestContextInjection:
             result = await client.call_tool("tool_with_resource", {})
             assert len(result.content) == 1
             assert "Read resource: resource data" in result.content[0].text
+
+
+class TestServerPrompts:
+    """Test prompt functionality in FastMCP server."""
+
+    async def test_prompt_decorator(self):
+        """Test that the prompt decorator registers prompts correctly."""
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def fn() -> str:
+            return "Hello, world!"
+
+        prompts = mcp._prompt_manager.list_prompts()
+        assert len(prompts) == 1
+        assert prompts[0].name == "fn"
+        # Don't compare functions directly since validate_call wraps them
+        assert await prompts[0].render() == [
+            UserMessage(content=TextContent(type="text", text="Hello, world!"))
+        ]
+
+    def test_prompt_decorator_with_name(self):
+        """Test prompt decorator with custom name."""
+        mcp = FastMCP()
+
+        @mcp.prompt(name="custom")
+        def fn() -> str:
+            return "Hello, world!"
+
+        prompts = mcp._prompt_manager.list_prompts()
+        assert len(prompts) == 1
+        assert prompts[0].name == "custom"
+
+    def test_prompt_decorator_with_description(self):
+        """Test prompt decorator with custom description."""
+        mcp = FastMCP()
+
+        @mcp.prompt(description="A custom description")
+        def fn() -> str:
+            return "Hello, world!"
+
+        prompts = mcp._prompt_manager.list_prompts()
+        assert len(prompts) == 1
+        assert prompts[0].description == "A custom description"
+
+    def test_prompt_decorator_error(self):
+        """Test error when decorator is used incorrectly."""
+        mcp = FastMCP()
+        with pytest.raises(TypeError, match="decorator was used incorrectly"):
+
+            @mcp.prompt
+            def fn() -> str:
+                return "Hello, world!"
+
+    async def test_list_prompts(self):
+        """Test listing prompts through MCP protocol."""
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def fn(name: str, optional: str = "default") -> str:
+            return f"Hello, {name}!"
+
+        async with client_session(mcp._mcp_server) as client:
+            result = await client.list_prompts()
+            assert len(result.prompts) == 1
+            assert result.prompts[0].name == "fn"
+            assert len(result.prompts[0].arguments) == 2
+            assert result.prompts[0].arguments[0].name == "name"
+            assert result.prompts[0].arguments[0].required is True
+            assert result.prompts[0].arguments[1].name == "optional"
+            assert result.prompts[0].arguments[1].required is False
+
+    async def test_get_prompt(self):
+        """Test getting a prompt through MCP protocol."""
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def fn(name: str) -> str:
+            return f"Hello, {name}!"
+
+        async with client_session(mcp._mcp_server) as client:
+            result = await client.get_prompt("fn", {"name": "World"})
+            assert len(result.messages) == 1
+            assert result.messages[0].role == "user"
+            assert result.messages[0].content.type == "text"
+            assert result.messages[0].content.text == "Hello, World!"
+
+    async def test_get_prompt_with_resource(self):
+        """Test getting a prompt that returns resource content."""
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def fn() -> Message:
+            return UserMessage(
+                content=EmbeddedResource(
+                    type="resource",
+                    resource={
+                        "uri": "file://test.txt",
+                        "text": "File contents",
+                        "mimeType": "text/plain",
+                    },
+                )
+            )
+
+        async with client_session(mcp._mcp_server) as client:
+            result = await client.get_prompt("fn")
+            assert len(result.messages) == 1
+            assert result.messages[0].role == "user"
+            assert result.messages[0].content.type == "resource"
+            assert str(result.messages[0].content.resource.uri) == "file://test.txt/"
+            assert result.messages[0].content.resource.text == "File contents"
+            assert result.messages[0].content.resource.mimeType == "text/plain"
+
+    async def test_get_unknown_prompt(self):
+        """Test error when getting unknown prompt."""
+        mcp = FastMCP()
+        async with client_session(mcp._mcp_server) as client:
+            with pytest.raises(McpError, match="Unknown prompt"):
+                await client.get_prompt("unknown")
+
+    async def test_get_prompt_missing_args(self):
+        """Test error when required arguments are missing."""
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def fn(name: str) -> str:
+            return f"Hello, {name}!"
+
+        async with client_session(mcp._mcp_server) as client:
+            with pytest.raises(McpError, match="Missing required arguments"):
+                await client.get_prompt("fn")

uv.lock

@@ -231,7 +231,7 @@ wheels = [
 
 [[package]]
 name = "fastmcp"
-version = "0.2.1.dev8+gfc560b3.d20241130"
+version = "0.2.1.dev44+gbfae276.d20241130"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },
@@ -254,8 +254,8 @@ dev = [
 [package.metadata]
 requires-dist = [
     { name = "httpx", specifier = ">=0.26.0" },
-    { name = "mcp", specifier = ">=1.0.0" },
-    { name = "pydantic", specifier = ">=2.5.3" },
+    { name = "mcp", specifier = ">=1.0.0,<2.0.0" },
+    { name = "pydantic", specifier = ">=2.5.3,<3.0.0" },
     { name = "pydantic-settings", specifier = ">=2.6.1" },
     { name = "typer", specifier = ">=0.9.0" },
 ]