diff --git a/mirascope/mcp/__init__.py b/mirascope/mcp/__init__.py
new file mode 100644
index 000000000..ccdae5170
--- /dev/null
+++ b/mirascope/mcp/__init__.py
@@ -0,0 +1,6 @@
+"""Mirascope Model Context Protocol (MCP) implementation."""
+
+from .server import MCPServer
+from .tools import MCPTool
+
+__all__ = ["MCPServer", "MCPTool"]
diff --git a/mirascope/mcp/server.py b/mirascope/mcp/server.py
new file mode 100644
index 000000000..72308447a
--- /dev/null
+++ b/mirascope/mcp/server.py
@@ -0,0 +1,343 @@
+"""MCP server implementation."""
+
+import inspect
+from collections.abc import Awaitable, Callable, Iterable
+from typing import Literal, ParamSpec, cast, overload
+
+import mcp.server.stdio
+from docstring_parser import parse
+from mcp.server import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+from mcp.types import (
+    GetPromptResult,
+    ImageContent,
+    Prompt,
+    PromptArgument,
+    PromptMessage,
+    Resource,
+    TextContent,
+    Tool,
+)
+from pydantic import AnyUrl, BaseModel
+
+from mirascope.core import BaseDynamicConfig, BaseMessageParam, BaseTool
+from mirascope.core.base import ImagePart, TextPart
+from mirascope.core.base._utils import (
+    MessagesDecorator,
+    convert_base_model_to_base_tool,
+    convert_function_to_base_tool,
+    fn_is_async,
+)
+from mirascope.core.base._utils._messages_decorator import (
+    MessagesAsyncFunction,
+    MessagesSyncFunction,
+)
+from mirascope.core.base.prompt import PromptDecorator, prompt_template
+from mirascope.mcp.tools import MCPTool, ToolUseBlock
+
+_P = ParamSpec("_P")
+
+
+def _convert_base_message_param_to_prompt_messages(
+    base_message_param: BaseMessageParam,
+) -> list[PromptMessage]:
+    """
+    Convert BaseMessageParam to types.PromptMessage.
+
+    Args:
+        base_message_param: BaseMessageParam instance.
+
+    Returns:
+        A list of types.PromptMessage instances.
+    """
+
+    # Validate role
+    role = base_message_param.role
+    if role not in ["user", "assistant"]:
+        raise ValueError(f"invalid role: {role}")
+
+    if isinstance(base_message_param.content, str):
+        contents = [TextContent(type="text", text=base_message_param.content)]
+    elif isinstance(base_message_param.content, Iterable):
+        contents = []
+        for part in base_message_param.content:
+            if isinstance(part, TextPart):
+                contents.append(TextContent(type="text", text=part.text))
+            elif isinstance(part, ImagePart):
+                contents.append(
+                    ImageContent(
+                        type="image",
+                        data=part.image.decode("utf-8"),
+                        mimeType=part.media_type,
+                    )
+                )
+            else:
+                raise ValueError(f"Unsupported content type: {type(part)}")
+    else:
+        raise ValueError(
+            f"Unsupported content type: {type(base_message_param.content)}"
+        )
+
+    return [
+        PromptMessage(role=cast(Literal["user", "assistant"], role), content=content)
+        for content in contents
+    ]
+
+
+def _generate_prompt_from_function(fn: Callable) -> Prompt:
+    """
+    Generate a Prompt object from a function, extracting metadata like argument descriptions
+    from the function's docstring and type hints.
+
+    Args:
+        fn (Callable): The function to process.
+
+    Returns:
+        Prompt: A structured Prompt object with metadata.
+    """
+    # Parse the docstring for structured information
+    docstring = parse(fn.__doc__ or "")
+
+    # Extract general description from the docstring
+    description = docstring.short_description or ""
+
+    # Prepare to extract argument details
+    signature = inspect.signature(fn)
+    parameter_docs = {param.arg_name: param.description for param in docstring.params}
+
+    arguments = []
+    for parameter in signature.parameters.values():
+        arg_desc = parameter_docs.get(parameter.name, "")
+        required = parameter.default is inspect.Parameter.empty
+        arguments.append(
+            PromptArgument(name=parameter.name, description=arg_desc, required=required)
+        )
+
+    return Prompt(name=fn.__name__, description=description, arguments=arguments)
+
+
+class MCPServer:
+    """MCP server implementation."""
+
+    def __init__(
+        self,
+        name: str,
+        version: str = "1.0.0",
+        tools: list[Callable | type[BaseTool]] | None = None,
+        resources: list[tuple[Resource, Callable]] | None = None,
+        prompts: list[
+            Callable[
+                ...,
+                list[BaseMessageParam]
+                | Callable[..., Awaitable[list[BaseMessageParam]]],
+            ]
+        ]
+        | None = None,
+    ) -> None:
+        self.name: str = name
+        self.version: str = version
+        self.server: Server = Server(name)
+        self._tools: dict[str, tuple[Tool, type[MCPTool]]] = {}
+        self._resources: dict[str, tuple[Resource, Callable]] = {}
+        self._prompts: dict[
+            str,
+            tuple[
+                Prompt,
+                Callable[..., Awaitable[list[BaseMessageParam]]]
+                | Callable[..., list[BaseMessageParam]],
+            ],
+        ] = {}
+        if tools:
+            for tool in tools:
+                self.tool()(tool)
+        if resources:
+            for resource, resource_func in resources:
+                self.resource(
+                    uri=str(resource.uri),
+                    name=resource.name,
+                    description=resource.description,
+                    mime_type=resource.mimeType,
+                )(resource_func)
+        if prompts:
+            for prompt in prompts:
+                self._register_prompt(prompt)
+
+    def tool(
+        self,
+    ) -> Callable[[Callable | type[BaseTool]], type[BaseTool]]:
+        """Decorator to register tools."""
+
+        def decorator(
+            tool: Callable | type[BaseTool],
+        ) -> type[BaseTool]:
+            if inspect.isclass(tool):
+                if issubclass(tool, MCPTool):
+                    converted_tool = tool
+                else:
+                    converted_tool = convert_base_model_to_base_tool(
+                        cast(type[BaseModel], tool), MCPTool
+                    )
+            else:
+                converted_tool = convert_function_to_base_tool(tool, MCPTool)
+            tool_schema = converted_tool.tool_schema()
+            name = tool_schema["name"]
+            if name in self._tools:
+                # Raise KeyError if tool name already exists
+                raise KeyError(f"Tool {name} already exists.")
+
+            self._tools[name] = (
+                Tool(
+                    name=name,
+                    description=tool_schema.get("description"),
+                    inputSchema=tool_schema["input_schema"],
+                ),
+                converted_tool,
+            )
+            return converted_tool
+
+        return decorator
+
+    def resource(
+        self,
+        uri: str,
+        name: str | None = None,
+        description: str | None = None,
+        mime_type: str | None = None,
+    ) -> Callable[[Callable], Callable]:
+        """Decorator to register resources."""
+
+        def decorator(func: Callable) -> Callable:
+            # Normalize URI so it always ends with a slash if needed
+            uri_with_slash = uri + "/" if not uri.endswith("/") else uri
+
+            if uri_with_slash in self._resources:
+                # Raise KeyError if resource URI already exists
+                raise KeyError(f"Resource {uri_with_slash} already exists.")
+
+            resource = Resource(
+                uri=cast(AnyUrl, uri_with_slash),
+                name=name or func.__name__,
+                mimeType=mime_type,
+                description=description
+                or parse(func.__doc__ or "").short_description
+                or "",
+            )
+            self._resources[str(resource.uri)] = resource, func
+            return func
+
+        return decorator
+
+    def _register_prompt(self, decorated_func: Callable) -> None:
+        prompt = _generate_prompt_from_function(decorated_func)
+        name = prompt.name
+        if name in self._prompts:
+            # Raise KeyError if prompt name already exists
+            raise KeyError(f"Prompt {name} already exists.")
+
+        self._prompts[prompt.name] = prompt, decorated_func
+
+    @overload
+    def prompt(self, template: str) -> PromptDecorator: ...
+
+    @overload
+    def prompt(self, template: None = None) -> MessagesDecorator: ...
+
+    def prompt(
+        self,
+        template: str | None = None,
+    ) -> PromptDecorator | MessagesDecorator:
+        """Decorator to register prompts."""
+
+        def decorator(
+            func: MessagesSyncFunction | MessagesAsyncFunction,
+        ) -> (
+            Callable[..., Awaitable[list[BaseMessageParam] | BaseDynamicConfig]]
+            | Callable[..., list[BaseMessageParam] | BaseDynamicConfig]
+        ):
+            decorated_prompt = prompt_template(template)(func)
+            self._register_prompt(decorated_prompt)
+            return decorated_prompt
+
+        return decorator  # pyright: ignore [reportReturnType]
+
+    async def run(self) -> None:
+        """Run the MCP server."""
+
+        @self.server.read_resource()
+        async def handle_read_resource(uri: AnyUrl) -> str:
+            resource_and_func = self._resources.get(str(uri))
+            if resource_and_func is None:
+                raise ValueError(f"Unknown resource: {uri}")
+
+            resource, func = resource_and_func
+            if fn_is_async(func):
+                ret = await func()
+            else:
+                ret = func()
+            return ret
+
+        @self.server.list_resources()
+        async def handle_list_resource() -> list[Resource]:
+            return [resource for resource, _ in self._resources.values()]
+
+        @self.server.list_tools()
+        async def list_tools() -> list[Tool]:
+            return [tool for tool, _ in self._tools.values()]
+
+        @self.server.call_tool()
+        async def call_tool(name: str, arguments: dict) -> list[TextContent]:
+            if name not in self._tools:
+                raise KeyError(f"Tool {name} not found.")
+            _, tool_type = self._tools[name]
+
+            tool = tool_type.from_tool_call(
+                tool_call=ToolUseBlock(id=name, name=name, input=arguments)
+            )
+            if fn_is_async(tool.call):
+                result = await tool.call()
+            else:
+                result = tool.call()
+            return [TextContent(type="text", text=result)]
+
+        @self.server.list_prompts()
+        async def handle_list_prompts() -> list[Prompt]:
+            return [prompt for prompt, _ in self._prompts.values()]
+
+        @self.server.get_prompt()
+        async def handle_get_prompt(
+            name: str, arguments: dict[str, str] | None
+        ) -> GetPromptResult:
+            if name not in self._prompts:
+                raise ValueError(f"Unknown prompt: {name}")
+            if arguments is None:
+                arguments = {}
+            prompt, decorated_func = self._prompts[name]
+            if fn_is_async(decorated_func):
+                messages: list[BaseMessageParam] = await decorated_func(**arguments)
+            else:
+                messages = decorated_func(**arguments)
+
+            return GetPromptResult(
+                description=f"{prompt.name} template for {arguments}",
+                messages=[
+                    prompt_message
+                    for message in messages
+                    for prompt_message in _convert_base_message_param_to_prompt_messages(
+                        message
+                    )
+                ],
+            )
+
+        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+            await self.server.run(
+                read_stream,
+                write_stream,
+                InitializationOptions(
+                    server_name=self.name,
+                    server_version=self.version,
+                    capabilities=self.server.get_capabilities(
+                        notification_options=NotificationOptions(),
+                        experimental_capabilities={},
+                    ),
+                ),
+            )
diff --git a/mirascope/mcp/tools.py b/mirascope/mcp/tools.py
new file mode 100644
index 000000000..397b13c02
--- /dev/null
+++ b/mirascope/mcp/tools.py
@@ -0,0 +1,94 @@
+"""The `MCPTool` class for easy tool usage with MCPTool LLM calls.
+
+usage docs: learn/tools.md
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+from pydantic.json_schema import SkipJsonSchema
+from typing_extensions import Required, TypedDict
+
+from mirascope.core import BaseTool
+from mirascope.core.base import ToolConfig
+
+
+class ToolParam(TypedDict, total=False):
+    input_schema: Required[dict[str, Any]]
+    name: Required[str]
+    description: str
+
+
+class ToolUseBlock(BaseModel):
+    id: str
+    input: dict[str, Any]
+    name: str
+
+
+class MCPToolToolConfig(ToolConfig, total=False):
+    """A tool configuration for mcp-specific features."""
+
+
+class MCPTool(BaseTool):
+    """A class for defining tools for MCP LLM calls.
+
+    Example:
+
+    ```python
+    import asyncio
+
+    from mirascope.mcp import MCPServer
+
+    app = MCPServer("book-recommend")
+
+
+    @app.tool()
+    async def recommend_book(genre: str) -> str:
+        return f"Recommend a {genre} book"
+
+    asyncio.run(app.run())
+    ```
+    """
+
+    __provider__ = "mcp"
+    __tool_config_type__ = MCPToolToolConfig
+
+    tool_call: SkipJsonSchema[ToolUseBlock]
+
+    @classmethod
+    def tool_schema(cls) -> ToolParam:
+        """Constructs a `ToolParam` tool schema from the `BaseModel` schema defined.
+
+        Example:
+        ```python
+        from mirascope.mcp import MCPTool
+
+
+        def format_book(title: str, author: str) -> str:
+            return f"{title} by {author}"
+
+
+        tool_type = MCPTool.type_from_fn(format_book)
+        print(tool_type.tool_schema())  # prints the MCP-specific tool schema
+        ```
+        """
+        kwargs = {
+            "input_schema": cls.model_json_schema(),
+            "name": cls._name(),
+            "description": cls._description(),
+        }
+        return ToolParam(**kwargs)
+
+    @classmethod
+    def from_tool_call(cls, tool_call: ToolUseBlock) -> MCPTool:
+        """Constructs an `MCPTool` instance from a `tool_call`.
+
+        Args:
+            tool_call: The MCP tool call from which to construct this tool
+                instance.
+        """
+        model_json = {"tool_call": tool_call}
+        model_json |= tool_call.input
+        return cls.model_validate(model_json)
diff --git a/pyproject.toml b/pyproject.toml
index 76cff7115..74c6b5fcf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -83,6 +83,9 @@ realtime = [
     "sounddevice>=0.5.1,<1",
     "websockets>=13.1,<14",
 ]
+mcp = [
+    "mcp>=1.0.0",
+]
 
 [tool.uv]
 dev-dependencies = [
diff --git a/tests/mcp/__init__.py b/tests/mcp/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/mcp/test_mcp_server.py b/tests/mcp/test_mcp_server.py
new file mode 100644
index 000000000..6b175ad2f
--- /dev/null
+++ b/tests/mcp/test_mcp_server.py
@@ -0,0 +1,1079 @@
+"""Tests for MCP server implementation."""
+
+import asyncio
+from typing import cast
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from mcp.types import Resource, Tool
+from pydantic import AnyUrl, BaseModel
+
+from mirascope.core import BaseMessageParam, prompt_template
+from mirascope.core.anthropic import AnthropicTool
+from mirascope.core.base import BaseTool, ImagePart, TextPart
+from mirascope.mcp import MCPTool
+from mirascope.mcp.server import (
+    MCPServer,
+    _convert_base_message_param_to_prompt_messages,
+    _generate_prompt_from_function,
+)
+
+
+# Add fixtures for common test data
+@pytest.fixture
+def sample_tool():
+    def recommend_book(title: str, author: str):
+        """Recommend a book with title and author.
+
+        Args:
+            title: Book title
+            author: Book author
+        """
+
+    return recommend_book
+
+
+def test_server_initialization():
+    """Test server initialization with different configurations."""
+
+    # Test with tools
+    def recommend_fantasy_book():
+        """Recommends a fantasy book."""
+
+    class FantasyBook(BaseModel):
+        """Fantasy book model."""
+
+        title: str
+        author: str
+
+    class FantasyBookTool(AnthropicTool):
+        """Fantasy book tool."""
+
+        title: str
+        author: str
+
+        def call(self) -> str: ...
+
+    # Test initializing with different tool types
+    server = MCPServer(
+        "book-recommend",
+        tools=[recommend_fantasy_book, FantasyBook, FantasyBookTool],
+        version="2.0.0",
+    )
+    assert len(server._tools) == 3
+    assert "recommend_fantasy_book" in server._tools
+
+    # Test with resources
+    resource = Resource(
+        uri=cast(AnyUrl, "file://data.txt/"),
+        name="Fantasy Books Database",
+        mimeType="text/plain",
+        description="Database of fantasy books",
+    )
+
+    async def read_data(): ...
+
+    server = MCPServer("book-recommend", resources=[(resource, read_data)])
+    assert len(server._resources) == 1
+    assert "file://data.txt/" in server._resources
+
+
+@pytest.mark.asyncio
+async def test_list_resources():
+    """Test listing resources."""
+    server = MCPServer("book-recommend")
+
+    @server.resource(
+        uri="file://fantasy-books.txt/", name="Fantasy Books", mime_type="text/plain"
+    )
+    async def read_fantasy_books(): ...
+
+    # Mock list_resources method
+    mock_list_resources = AsyncMock(
+        return_value=[
+            Resource(
+                uri=cast(AnyUrl, "file://fantasy-books.txt/"),
+                name="Fantasy Books",
+                mimeType="text/plain",
+            )
+        ]
+    )
+    server.server.list_resources = mock_list_resources
+
+    resources = await server.server.list_resources()
+    assert len(resources) == 1
+    assert resources[0].name == "Fantasy Books"
+
+
+@pytest.mark.asyncio
+async def test_list_tools():
+    """Test listing tools."""
+    server = MCPServer("book-recommend")
+
+    @server.tool()
+    def fantasy_book_search(title: str) -> str:  # pyright: ignore [reportReturnType]
+        """Search for fantasy books."""
+
+    # Mock list_tools method
+    mock_list_tools = AsyncMock(
+        return_value=[
+            Tool(
+                name="fantasy_book_search",
+                description="Search for fantasy books.",
+                inputSchema={"type": "object"},
+            )
+        ]
+    )
+    server.server.list_tools = mock_list_tools
+
+    tools = await server.server.list_tools()
+    assert len(tools) == 1
+    assert tools[0].name == "fantasy_book_search"
+
+
+@pytest.mark.asyncio
+async def test_async_prompt():
+    """Test async prompt handling."""
+    server = MCPServer("book-recommend")
+
+    @server.prompt()
+    async def recommend_fantasy_book(subgenre: str):
+        """Get fantasy book recommendations."""
+
+    # Mock get_prompt
+    mock_get_prompt = AsyncMock()
+    server.server.get_prompt = mock_get_prompt
+
+    await server.server.get_prompt()
+    mock_get_prompt.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_async_resource():
+    """Test async resource handling."""
+    server = MCPServer("book-recommend")
+
+    @server.resource(
+        uri="file://fantasy-books.txt/",
+        name="Fantasy Books",
+        description="Fantasy book database",
+    )
+    async def read_fantasy_books(): ...
+
+    # Mock read_resource
+    mock_read_resource = AsyncMock()
+    server.server.read_resource = mock_read_resource
+
+    await server.server.read_resource()
+    mock_read_resource.assert_called_once()
+
+
+def test_convert_base_message_param():
+    """Test conversion of BaseMessageParam to PromptMessage."""
+    # Test text content
+    text_message = BaseMessageParam(role="user", content="Recommend a fantasy book")
+    prompt_messages = _convert_base_message_param_to_prompt_messages(text_message)
+    assert len(prompt_messages) == 1
+    assert prompt_messages[0].role == "user"  # pyright: ignore [reportAttributeAccessIssue]
+    assert prompt_messages[0].content.text == "Recommend a fantasy book"  # pyright: ignore [reportAttributeAccessIssue]
+
+    # Test multiple content parts
+    multi_content = [
+        TextPart(type="text", text="Looking for fantasy books"),
+        TextPart(type="text", text="with dragons"),
+    ]
+    multi_message = BaseMessageParam(role="user", content=multi_content)
+    prompt_messages = _convert_base_message_param_to_prompt_messages(multi_message)
+    assert len(prompt_messages) == 2
+
+    # Test invalid role
+    with pytest.raises(ValueError, match="invalid role"):
+        _convert_base_message_param_to_prompt_messages(
+            BaseMessageParam(role="invalid", content="test")
+        )
+
+
+def test_tool_decorator_with_different_types(sample_tool):
+    """Test tool decorator with different input types."""
+    server = MCPServer("book-recommend")
+
+    # Test with function
+    tool_fn = server.tool()(sample_tool)
+    assert tool_fn.__name__ == "recommend_book"
+
+    # Test with BaseModel
+    class BookModel(BaseModel):
+        title: str
+        author: str
+
+    tool_model = server.tool()(BookModel)
+    assert "title" in tool_model.model_fields
+    assert "author" in tool_model.model_fields
+
+    # Test with AnthropicTool
+    class BookTool(AnthropicTool):
+        title: str
+        author: str
+
+        def call(self) -> str: ...
+
+    tool_anthropic = server.tool()(BookTool)
+    assert issubclass(tool_anthropic, AnthropicTool)
+
+
+def test_prompt_message_conversion():
+    """Test conversion of various message content types."""
+    # Test text content with multiple parts
+    message = BaseMessageParam(
+        role="user",
+        content=[
+            TextPart(type="text", text="Part 1"),
+            TextPart(type="text", text="Part 2"),
+        ],
+    )
+    result = _convert_base_message_param_to_prompt_messages(message)
+    assert len(result) == 2
+    assert result[0].content.text == "Part 1"  # pyright: ignore [reportAttributeAccessIssue]
+    assert result[1].content.text == "Part 2"  # pyright: ignore [reportAttributeAccessIssue]
+
+
+def test_convert_base_message_param_to_prompt_messages_image_part():
+    image_data = b"fakeimagebytes"
+    # Provide all required fields for ImagePart
+    part = ImagePart(
+        type="image", detail="image detail", image=image_data, media_type="image/png"
+    )
+    message = BaseMessageParam(role="assistant", content=[part])
+    result = _convert_base_message_param_to_prompt_messages(message)
+    assert len(result) == 1
+    assert result[0].role == "assistant"
+    assert result[0].content.type == "image"
+    assert result[0].content.mimeType == "image/png"
+    assert result[0].content.data == image_data.decode("utf-8")
+
+
+def test_generate_prompt_from_function_no_docstring():
+    # function with no docstring
+    def no_doc_func(x: str): ...
+
+    prompt = _generate_prompt_from_function(no_doc_func)
+    assert prompt.name == "no_doc_func"
+    # no description
+    assert prompt.description == ""
+    assert isinstance(prompt.arguments, list)
+    assert len(prompt.arguments) == 1
+    assert prompt.arguments[0].name == "x"
+    assert prompt.arguments[0].description == ""
+
+
+def test_generate_prompt_from_function_with_docstring():
+    # function with docstring and param desc
+    def func_with_docs(x: str, y: int):
+        """This is a test function.
+
+        Args:
+            x: The X parameter.
+            y: The Y parameter.
+        """
+
+    prompt = _generate_prompt_from_function(func_with_docs)
+    assert prompt.name == "func_with_docs"
+    assert prompt.description == "This is a test function."
+    assert isinstance(prompt.arguments, list)
+    assert len(prompt.arguments) == 2
+    arg_map = {arg.name: arg for arg in prompt.arguments}
+    assert arg_map["x"].description == "The X parameter."
+    assert arg_map["y"].description == "The Y parameter."
+
+
+def test_tool_decorator_name_conflict():
+    server = MCPServer("test")
+
+    @server.tool()
+    def mytool(a: str) -> str: ...
+
+    with pytest.raises(KeyError):
+        # same tool name again to force a conflict
+        @server.tool()
+        def mytool(a: str) -> str: ...
+
+
+def test_resource_decorator_name_conflict():
+    server = MCPServer("test")
+
+    @server.resource(uri="file://test.txt")
+    def read_res(): ...
+
+    with pytest.raises(KeyError):
+        # same resource URI again
+        @server.resource(uri="file://test.txt")
+        def read_res2(): ...
+
+
+def test_prompt_decorator_conflict():
+    server = MCPServer("test")
+
+    @server.prompt()
+    def prompt_func(x: str): ...  # pyright: ignore [reportRedeclaration]
+
+    with pytest.raises(KeyError):
+        # same prompt name (prompt_func)
+        @server.prompt()
+        def prompt_func(y: int): ...
+
+
+def test_resource_docstring():
+    server = MCPServer("test")
+
+    @server.resource(uri="file://docstring.txt")
+    def read_with_doc():
+        """Short description."""
+
+    # The server normalizes the URI by adding a slash
+    assert "file://docstring.txt/" in server._resources
+    resource, _ = server._resources["file://docstring.txt/"]
+    assert resource.description == "Short description."
+
+
+def test_tool_decorator_with_base_model():
+    class MyToolModel(BaseModel):
+        name: str
+        age: int
+
+    server = MCPServer("test")
+
+    tool_cls = server.tool()(MyToolModel)
+    assert issubclass(tool_cls, BaseTool)
+    # Check that tool was registered
+    assert "MyToolModel" in server._tools
+
+
+def test_tool_decorator_with_base_tool():
+    class MyBaseTool(BaseTool):
+        def call(self) -> str: ...
+
+    server = MCPServer("test")
+    tool_cls = server.tool()(MyBaseTool)
+    assert issubclass(tool_cls, BaseTool)
+    # Check registration
+    assert "MyBaseTool" in server._tools
+
+
+def test_tool_decorator_with_mcp_tool():
+    class MyToolModel(BaseModel):
+        name: str
+        age: int
+
+    MyMCPTool = MCPTool.type_from_base_model_type(MyToolModel)
+    server = MCPServer("test")
+    tool_cls = server.tool()(MyMCPTool)
+    assert issubclass(tool_cls, MCPTool)
+    # Check registration
+    assert "MyToolModel" in server._tools
+
+
+def test_prompt_decorator_with_template():
+    server = MCPServer("test")
+
+    @server.prompt("Recommend a {genre} book")
+    def prompt_func(genre: str): ...
+
+    assert "prompt_func" in server._prompts
+
+    prompt_data, _ = server._prompts["prompt_func"]
+    assert prompt_data.name == "prompt_func"
+    assert isinstance(prompt_data.arguments, list)
+    assert any(arg.name == "genre" for arg in prompt_data.arguments)
+
+
+@pytest.mark.asyncio
+async def test_list_resources_decorator():
+    handler_list_resources = None
+
+    def capture_list_resources():
+        def decorator(func):
+            nonlocal handler_list_resources
+            handler_list_resources = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.list_resources.side_effect = capture_list_resources
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.tool()
+        def get_book(genre: str) -> str: ...
+
+        @app.resource(
+            uri="file://fantasy-books.txt/",
+            name="Fantasy Books",
+            mime_type="text/plain",
+        )
+        async def read_data(): ...
+
+        @app.prompt()
+        def recommend_book(genre: str) -> str: ...
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        # Check if list_resources handler was captured
+        assert handler_list_resources is not None, "list_resources handler not captured"
+        resources_list = await handler_list_resources()  # pyright: ignore [reportGeneralTypeIssues]
+        assert isinstance(
+            resources_list, list
+        ), "Expected list from list_resources handler"
+
+
+@pytest.mark.asyncio
+async def test_read_resource_decorator():
+    handler_read_resource = None
+
+    def capture_read_resource():
+        def decorator(func):
+            nonlocal handler_read_resource
+            handler_read_resource = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.read_resource.side_effect = capture_read_resource
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.resource(
+            uri="file://test-resource/", name="Test Resource", mime_type="text/plain"
+        )
+        def read_test():
+            return "Test Resource Content"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_read_resource is not None, "read_resource handler not captured"
+        content = await handler_read_resource("file://test-resource/")  # pyright: ignore [reportGeneralTypeIssues]
+        assert content == "Test Resource Content"
+
+
+@pytest.mark.asyncio
+async def test_list_tools_decorator():
+    handler_list_tools = None
+
+    def capture_list_tools():
+        def decorator(func):
+            nonlocal handler_list_tools
+            handler_list_tools = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.list_tools.side_effect = capture_list_tools
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.tool()
+        def sample_tool(x: str) -> str: ...
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_list_tools is not None, "list_tools handler not captured"
+        tools_list = await handler_list_tools()  # pyright: ignore [reportGeneralTypeIssues]
+        assert isinstance(tools_list, list), "Expected list from list_tools handler"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_decorator():
+    handler_call_tool = None
+
+    def capture_call_tool():
+        def decorator(func):
+            nonlocal handler_call_tool
+            handler_call_tool = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.call_tool.side_effect = capture_call_tool
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.tool()
+        def sample_tool(x: str) -> str:
+            return f"Result for {x}"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_call_tool is not None, "call_tool handler not captured"
+        tool_result = await handler_call_tool("sample_tool", {"x": "test"})  # pyright: ignore [reportGeneralTypeIssues]
+        assert isinstance(tool_result, list), "Expected list from call_tool handler"
+
+
+@pytest.mark.asyncio
+async def test_list_prompts_decorator():
+    handler_list_prompts = None
+
+    def capture_list_prompts():
+        def decorator(func):
+            nonlocal handler_list_prompts
+            handler_list_prompts = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.list_prompts.side_effect = capture_list_prompts
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.prompt()
+        def sample_prompt(x: str) -> str: ...
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_list_prompts is not None, "list_prompts handler not captured"
+        prompts_list = await handler_list_prompts()  # pyright: ignore [reportGeneralTypeIssues]
+        assert isinstance(prompts_list, list), "Expected list from list_prompts handler"
+
+
+@pytest.mark.asyncio
+async def test_get_prompt_decorator():
+    handler_get_prompt = None
+
+    def capture_get_prompt():
+        def decorator(func):
+            nonlocal handler_get_prompt
+            handler_get_prompt = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.get_prompt.side_effect = capture_get_prompt
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("book-recommend")
+
+        @app.prompt()
+        def sample_prompt(genre: str) -> str:
+            return f"Prompt for {genre}"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_get_prompt is not None, "get_prompt handler not captured"
+        # Call the handler_get_prompt with some arguments
+        prompt_result = await handler_get_prompt("sample_prompt", {"genre": "fantasy"})  # pyright: ignore [reportGeneralTypeIssues]
+        # prompt_result should be a GetPromptResult or similar
+        prompt_dict = prompt_result.dict()
+        assert "messages" in prompt_dict, "get_prompt result should have messages"
+
+
+def test_convert_base_message_param_to_prompt_messages_valid_str_user():
+    # role: user, content: str
+    param = BaseMessageParam(role="user", content="Hello")
+    msgs = _convert_base_message_param_to_prompt_messages(param)
+    assert len(msgs) == 1
+    assert msgs[0].role == "user"  # pyright: ignore [reportAttributeAccessIssue]
+    assert msgs[0].content.text == "Hello"  # pyright: ignore [reportAttributeAccessIssue]
+
+
+def test_convert_base_message_param_to_prompt_messages_valid_str_assistant():
+    # role: assistant, content: str
+    param = BaseMessageParam(role="assistant", content="Hi there")
+    msgs = _convert_base_message_param_to_prompt_messages(param)
+    assert len(msgs) == 1
+    assert msgs[0].role == "assistant"  # pyright: ignore [reportAttributeAccessIssue]
+    assert msgs[0].content.text == "Hi there"  # pyright: ignore [reportAttributeAccessIssue]
+
+
+def test_convert_base_message_param_to_prompt_messages_valid_iterable_text():
+    # role: user, content: iterable of TextPart(type="text")
+    param = BaseMessageParam(
+        role="user",
+        content=[
+            TextPart(type="text", text="Part1"),
+            TextPart(type="text", text="Part2"),
+        ],
+    )
+    msgs = _convert_base_message_param_to_prompt_messages(param)
+    assert len(msgs) == 2
+    assert msgs[0].content.text == "Part1"  # pyright: ignore [reportAttributeAccessIssue]
+    assert msgs[1].content.text == "Part2"  # pyright: ignore [reportAttributeAccessIssue]
+
+
+def test_convert_base_message_param_to_prompt_messages_valid_iterable_image():
+    # role: assistant, content: iterable of ImagePart(type="image")
+    param = BaseMessageParam(
+        role="assistant",
+        content=[
+            ImagePart(
+                type="image",
+                detail="An image",
+                image=b"fakeimage",
+                media_type="image/png",
+            ),
+        ],
+    )
+    msgs = _convert_base_message_param_to_prompt_messages(param)
+    assert len(msgs) == 1
+    assert msgs[0].role == "assistant"
+    assert msgs[0].content.type == "image"
+    assert msgs[0].content.mimeType == "image/png"
+    assert msgs[0].content.data == "fakeimage"
+
+
+def test_convert_base_message_param_to_prompt_messages_invalid_role():
+    # invalid role
+    param = BaseMessageParam(role="invalid", content="Hello")
+    with pytest.raises(ValueError, match="invalid role"):
+        _convert_base_message_param_to_prompt_messages(param)
+
+
+def test_convert_base_message_param_to_prompt_messages_iterable_unknown_part():
+    class UnknownPart:
+        pass
+
+    param = BaseMessageParam(role="user", content=[TextPart(type="text", text="Valid")])
+    param.content = [UnknownPart()]  # pyright: ignore [reportAttributeAccessIssue]
+
+    with pytest.raises(
+        ValueError, match="Unsupported content type: <class '.*UnknownPart'>"
+    ):
+        _convert_base_message_param_to_prompt_messages(param)
+
+
+def test_convert_base_message_param_to_prompt_messages_non_str_non_iterable():
+    # content neither str nor iterable
+    param = BaseMessageParam(role="user", content="Valid String")
+    param.content = 12345  # pyright: ignore [reportAttributeAccessIssue]
+
+    with pytest.raises(ValueError, match="Unsupported content type: <class 'int'>"):
+        _convert_base_message_param_to_prompt_messages(param)
+
+
+@pytest.mark.asyncio
+async def test_constructor_with_prompts():
+    @prompt_template()
+    def recommend_book_prompt(genre: str) -> str:
+        """Return a string recommending a book for the given genre."""
+        return f"Recommend a {genre} book"
+
+    app = MCPServer("book-recommend", prompts=[recommend_book_prompt])
+
+    assert "recommend_book_prompt" in app._prompts
+
+    mock_read_stream = AsyncMock()
+    mock_write_stream = AsyncMock()
+
+    with patch("mcp.server.stdio.stdio_server") as mock_stdio:
+        mock_cm = AsyncMock()
+        mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+        mock_stdio.return_value = mock_cm
+
+        await app.run()
+
+    (prompt, prompt_func) = app._prompts["recommend_book_prompt"]
+    assert prompt.name == "recommend_book_prompt"
+    assert (
+        prompt.description == "Return a string recommending a book for the given genre."
+    )
+    assert isinstance(prompt.arguments, list)
+    assert len(prompt.arguments) == 1
+    assert prompt.arguments[0].name == "genre"
+
+    result = prompt_func("fantasy")
+    assert isinstance(result, list)
+    assert len(result) == 1
+    assert isinstance(result[0], BaseMessageParam)
+    assert result[0].content == "Recommend a fantasy book"
+
+
+@pytest.mark.asyncio
+async def test_read_resource_decorator_non_existent():
+    """Test reading a non-existent resource and expecting a ValueError."""
+    handler_read_resource = None
+
+    def capture_read_resource():
+        def decorator(func):
+            nonlocal handler_read_resource
+            handler_read_resource = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.read_resource.side_effect = capture_read_resource
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        # Create an MCPServer with no resources
+        app = MCPServer("test")
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_read_resource is not None, "read_resource handler not captured"
+
+        # Since no resources were defined, reading any URI should raise ValueError.
+        with pytest.raises(ValueError, match="Unknown resource"):
+            await handler_read_resource("file://nonexistent-resource/")  # pyright: ignore
+
+
+@pytest.mark.asyncio
+async def test_read_resource_decorator_async_function():
+    """Test reading a resource from an async function using the captured handler."""
+    handler_read_resource = None
+
+    def capture_read_resource():
+        def decorator(func):
+            nonlocal handler_read_resource
+            handler_read_resource = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.read_resource.side_effect = capture_read_resource
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("test")
+
+        @app.resource(uri="file://async-resource/")
+        async def async_resource():
+            # Simulate async work
+            await asyncio.sleep(0.01)
+            return "Async Resource Data"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_read_resource is not None, "read_resource handler not captured"
+
+        # Call the captured handler to read the async resource
+        content = await handler_read_resource("file://async-resource/")  # pyright: ignore
+        assert content == "Async Resource Data"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_non_existent_tool():
+    """Test calling a non-existent tool should raise KeyError."""
+    handler_call_tool = None
+
+    def capture_call_tool():
+        def decorator(func):
+            nonlocal handler_call_tool
+            handler_call_tool = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.call_tool.side_effect = capture_call_tool
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        # Create MCPServer with no tools
+        app = MCPServer("test-call-tool")
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_call_tool is not None, "call_tool handler not captured"
+
+        with pytest.raises(KeyError, match="Tool nonexistent_tool not found."):
+            await handler_call_tool("nonexistent_tool", {})  # pyright: ignore
+
+
+@pytest.mark.asyncio
+async def test_call_tool_async_function():
+    """Test calling a tool implemented as an async function."""
+    handler_call_tool = None
+
+    def capture_call_tool():
+        def decorator(func):
+            nonlocal handler_call_tool
+            handler_call_tool = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.call_tool.side_effect = capture_call_tool
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("test-async-tool")
+
+        @app.tool()
+        async def async_tool(x: str) -> str:
+            await asyncio.sleep(0.01)
+            return f"Async result for {x}"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_call_tool is not None, "call_tool handler not captured"
+
+        result = await handler_call_tool("async_tool", {"x": "test"})  # pyright: ignore
+        # Should return a list of TextContent
+        assert len(result) == 1
+        assert result[0].text == "Async result for test"
+
+
+@pytest.mark.asyncio
+async def test_get_prompt_non_existent_prompt():
+    """Test requesting a non-existent prompt should raise ValueError."""
+    handler_get_prompt = None
+
+    def capture_get_prompt():
+        def decorator(func):
+            nonlocal handler_get_prompt
+            handler_get_prompt = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.get_prompt.side_effect = capture_get_prompt
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        # Create MCPServer with no prompts
+        app = MCPServer("test-get-prompt")
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_get_prompt is not None, "get_prompt handler not captured"
+
+        with pytest.raises(ValueError, match="Unknown prompt: nonexistent_prompt"):
+            await handler_get_prompt("nonexistent_prompt", {})  # pyright: ignore
+
+
+@pytest.mark.asyncio
+async def test_get_prompt_async_function():
+    """Test requesting a prompt implemented as an async function."""
+    handler_get_prompt = None
+
+    def capture_get_prompt():
+        def decorator(func):
+            nonlocal handler_get_prompt
+            handler_get_prompt = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.get_prompt.side_effect = capture_get_prompt
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("test-async-prompt")
+
+        @app.prompt()
+        async def async_prompt(genre: str) -> list[BaseMessageParam]:
+            await asyncio.sleep(0.01)
+            return [BaseMessageParam(role="user", content=f"Async prompt for {genre}")]
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_get_prompt is not None, "get_prompt handler not captured"
+
+        prompt_result = await handler_get_prompt("async_prompt", {"genre": "fantasy"})  # pyright: ignore
+        # Should return a GetPromptResult with messages
+        prompt_dict = prompt_result.dict()
+        assert "messages" in prompt_dict
+        assert len(prompt_dict["messages"]) == 1
+        assert (
+            prompt_dict["messages"][0]["content"]["text"] == "Async prompt for fantasy"
+        )
+
+
+@pytest.mark.asyncio
+async def test_get_prompt_no_arguments():
+    """Test get_prompt handler when arguments is None."""
+    handler_get_prompt = None
+
+    def capture_get_prompt():
+        def decorator(func):
+            nonlocal handler_get_prompt
+            handler_get_prompt = func
+            return func
+
+        return decorator
+
+    with patch("mirascope.mcp.server.Server") as MockServer:
+        server_instance = MagicMock()
+        server_instance.get_prompt.side_effect = capture_get_prompt
+        server_instance.get_capabilities.return_value = {}
+        server_instance.run = AsyncMock(return_value=None)
+
+        MockServer.return_value = server_instance
+
+        app = MCPServer("test-get-prompt-none-arguments")
+
+        @app.prompt()
+        def sample_prompt() -> str:
+            """A prompt that recommends a book based on genre."""
+            return "Recommend a fantasy book"
+
+        mock_read_stream = AsyncMock()
+        mock_write_stream = AsyncMock()
+
+        with patch("mcp.server.stdio.stdio_server") as mock_stdio_server:
+            mock_cm = AsyncMock()
+            mock_cm.__aenter__.return_value = (mock_read_stream, mock_write_stream)
+            mock_stdio_server.return_value = mock_cm
+
+            await app.run()
+
+        assert handler_get_prompt is not None, "get_prompt handler not captured"
+        prompt_result = await handler_get_prompt("sample_prompt", None)  # pyright: ignore
+
+        prompt_dict = prompt_result.dict()
+        assert "messages" in prompt_dict, "get_prompt result should have messages"
+        assert len(prompt_dict["messages"]) == 1
+        assert (
+            prompt_dict["messages"][0]["content"]["text"] == "Recommend a fantasy book"
+        )
diff --git a/uv.lock b/uv.lock
index a47350a18..2bf40f890 100644
--- a/uv.lock
+++ b/uv.lock
@@ -224,7 +224,7 @@ wheels = [
 
 [[package]]
 name = "anyio"
-version = "4.4.0"
+version = "4.6.2.post1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "exceptiongroup", marker = "python_full_version < '3.11'" },
@@ -232,9 +232,9 @@ dependencies = [
     { name = "sniffio" },
     { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/e6/e3/c4c8d473d6780ef1853d630d581f70d655b4f8d7553c6997958c283039a2/anyio-4.4.0.tar.gz", hash = "sha256:5aadc6a1bbb7cdb0bede386cac5e2940f5e2ff3aa20277e991cf028e0585ce94", size = 163930 }
+sdist = { url = "https://files.pythonhosted.org/packages/9f/09/45b9b7a6d4e45c6bcb5bf61d19e3ab87df68e0601fa8c5293de3542546cc/anyio-4.6.2.post1.tar.gz", hash = "sha256:4c8bc31ccdb51c7f7bd251f51c609e038d63e34219b44aa86e47576389880b4c", size = 173422 }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7b/a2/10639a79341f6c019dedc95bd48a4928eed9f1d1197f4c04f546fc7ae0ff/anyio-4.4.0-py3-none-any.whl", hash = "sha256:c1b2d8f46a8a812513012e1107cb0e68c17159a7a594208005a57dc776e1bdc7", size = 86780 },
+    { url = "https://files.pythonhosted.org/packages/e4/f5/f2b75d2fc6f1a260f340f0e7c6a060f4dd2961cc16884ed851b0d18da06a/anyio-4.6.2.post1-py3-none-any.whl", hash = "sha256:6d170c36fba3bdd840c73d3868c1e777e33676a69c3a72cf0a0d5d6d8009b61d", size = 90377 },
 ]
 
 [[package]]
@@ -3092,6 +3092,23 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/8f/8e/9ad090d3553c280a8060fbf6e24dc1c0c29704ee7d1c372f0c174aa59285/matplotlib_inline-0.1.7-py3-none-any.whl", hash = "sha256:df192d39a4ff8f21b1895d72e6a13f5fcc5099f00fa84384e0ea28c2cc0653ca", size = 9899 },
 ]
 
+[[package]]
+name = "mcp"
+version = "1.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "httpx" },
+    { name = "httpx-sse" },
+    { name = "pydantic" },
+    { name = "sse-starlette" },
+    { name = "starlette" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/97/de/a9ec0a1b6439f90ea59f89004bb2e7ec6890dfaeef809751d9e6577dca7e/mcp-1.0.0.tar.gz", hash = "sha256:dba51ce0b5c6a80e25576f606760c49a91ee90210fed805b530ca165d3bbc9b7", size = 82891 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/56/89/900c0c8445ec001d3725e475fc553b0feb2e8a51be018f3bb7de51e683db/mcp-1.0.0-py3-none-any.whl", hash = "sha256:bbe70ffa3341cd4da78b5eb504958355c68381fb29971471cea1e642a2af5b8a", size = 36361 },
+]
+
 [[package]]
 name = "mdit-py-plugins"
 version = "0.4.2"
@@ -3139,7 +3156,7 @@ wheels = [
 
 [[package]]
 name = "mirascope"
-version = "1.11.1"
+version = "1.11.2"
 source = { editable = "." }
 dependencies = [
     { name = "docstring-parser" },
@@ -3184,6 +3201,9 @@ litellm = [
 logfire = [
     { name = "logfire" },
 ]
+mcp = [
+    { name = "mcp" },
+]
 mistral = [
     { name = "mistralai" },
 ]
@@ -3262,6 +3282,7 @@ requires-dist = [
     { name = "langfuse", marker = "extra == 'langfuse'", specifier = ">=2.30.0,<3" },
     { name = "litellm", marker = "extra == 'litellm'", specifier = ">=1.42.12,<2" },
     { name = "logfire", marker = "extra == 'logfire'", specifier = ">=1.0.0,<3" },
+    { name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.0.0" },
     { name = "mistralai", marker = "extra == 'mistral'", specifier = ">=1.0.0,<2" },
     { name = "numpy", marker = "extra == 'realtime'", specifier = ">=1.26.4,<2" },
     { name = "openai", marker = "extra == 'openai'", specifier = ">=1.6.0,<2" },
@@ -5891,6 +5912,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5d/a5/b2860373aa8de1e626b2bdfdd6df4355f0565b47e51f7d0c54fe70faf8fe/sqlparse-0.5.1-py3-none-any.whl", hash = "sha256:773dcbf9a5ab44a090f3441e2180efe2560220203dc2f8c0b0fa141e18b505e4", size = 44156 },
 ]
 
+[[package]]
+name = "sse-starlette"
+version = "2.1.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "starlette" },
+    { name = "uvicorn" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/72/fc/56ab9f116b2133521f532fce8d03194cf04dcac25f583cf3d839be4c0496/sse_starlette-2.1.3.tar.gz", hash = "sha256:9cd27eb35319e1414e3d2558ee7414487f9529ce3b3cf9b21434fd110e017169", size = 19678 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/52/aa/36b271bc4fa1d2796311ee7c7283a3a1c348bad426d37293609ca4300eef/sse_starlette-2.1.3-py3-none-any.whl", hash = "sha256:8ec846438b4665b9e8c560fcdea6bc8081a3abf7942faa95e5a744999d219772", size = 9383 },
+]
+
 [[package]]
 name = "stack-data"
 version = "0.6.3"