crewAI/lib/crewai-tools/src/crewai_tools/adapters/mcp_adapter.py

"""MCPServer for CrewAI."""

from __future__ import annotations

from collections.abc import Callable
import logging
from typing import TYPE_CHECKING, Any

from crewai.tools import BaseTool
from crewai.utilities.pydantic_schema_utils import create_model_from_schema
from crewai.utilities.string_utils import sanitize_tool_name
from pydantic import BaseModel

from crewai_tools.adapters.tool_collection import ToolCollection


if TYPE_CHECKING:
    from mcp import StdioServerParameters
    from mcp.types import CallToolResult, TextContent, Tool
    from mcpadapt.core import MCPAdapt, ToolAdapter


logger = logging.getLogger(__name__)


try:
    from mcp import StdioServerParameters
    from mcp.types import CallToolResult, TextContent, Tool
    from mcpadapt.core import MCPAdapt, ToolAdapter

    class CrewAIToolAdapter(ToolAdapter):
        """Adapter that creates CrewAI tools with properly normalized JSON schemas.

        This adapter bypasses mcpadapt's model creation which adds invalid null values
        to field schemas, instead using CrewAI's own schema utilities.
        """

        def adapt(
            self,
            func: Callable[[dict[str, Any] | None], CallToolResult],
            mcp_tool: Tool,
        ) -> BaseTool:
            """Adapt a MCP tool to a CrewAI tool.

            Args:
                func: The function to call when the tool is invoked.
                mcp_tool: The MCP tool definition to adapt.

            Returns:
                A CrewAI BaseTool instance.
            """
            tool_name = sanitize_tool_name(mcp_tool.name)
            tool_description = mcp_tool.description or ""
            args_model = create_model_from_schema(mcp_tool.inputSchema)

            class CrewAIMCPTool(BaseTool):
                name: str = tool_name
                description: str = tool_description
                args_schema: type[BaseModel] = args_model

                def _run(self, **kwargs: Any) -> Any:
                    result = func(kwargs)
                    if len(result.content) == 1:
                        first_content = result.content[0]
                        if isinstance(first_content, TextContent):
                            return first_content.text
                        return str(first_content)
                    return str(
                        [
                            content.text
                            for content in result.content
                            if isinstance(content, TextContent)
                        ]
                    )

                def _generate_description(self) -> None:
                    schema = self.args_schema.model_json_schema()
                    schema.pop("$defs", None)
                    self.description = (
                        f"Tool Name: {self.name}\n"
                        f"Tool Arguments: {schema}\n"
                        f"Tool Description: {self.description}"
                    )

            return CrewAIMCPTool()

        async def async_adapt(self, afunc: Any, mcp_tool: Tool) -> Any:
            """Async adaptation is not supported by CrewAI."""
            raise NotImplementedError("async is not supported by the CrewAI framework.")

    MCP_AVAILABLE = True
except ImportError as e:
    logger.debug(f"MCP packages not available: {e}")
    MCP_AVAILABLE = False


class MCPServerAdapter:
    """Manages the lifecycle of an MCP server and make its tools available to CrewAI.

    Note: tools can only be accessed after the server has been started with the
        `start()` method.

    Usage:
        # context manager + stdio
        with MCPServerAdapter(...) as tools:
            # tools is now available

        # context manager + sse
        with MCPServerAdapter({"url": "http://localhost:8000/sse"}) as tools:
            # tools is now available

        # context manager with filtered tools
        with MCPServerAdapter(..., "tool1", "tool2") as filtered_tools:
            # only tool1 and tool2 are available

        # context manager with custom connect timeout (60 seconds)
        with MCPServerAdapter(..., connect_timeout=60) as tools:
            # tools is now available with longer timeout

        # manually stop mcp server
        try:
            mcp_server = MCPServerAdapter(...)
            tools = mcp_server.tools  # all tools

            # or with filtered tools and custom timeout
            mcp_server = MCPServerAdapter(..., "tool1", "tool2", connect_timeout=45)
            filtered_tools = mcp_server.tools  # only tool1 and tool2
            ...
        finally:
            mcp_server.stop()

        # Best practice is ensure cleanup is done after use.
        mcp_server.stop() # run after crew().kickoff()
    """

    def __init__(
        self,
        serverparams: StdioServerParameters | dict[str, Any],
        *tool_names: str,
        connect_timeout: int = 30,
    ) -> None:
        """Initialize the MCP Server.

        Args:
            serverparams: The parameters for the MCP server it supports either a
                `StdioServerParameters` or a `dict` respectively for STDIO and SSE.
            *tool_names: Optional names of tools to filter. If provided, only tools with
                matching names will be available.
            connect_timeout: Connection timeout in seconds to the MCP server (default is 30s).

        """
        super().__init__()
        self._adapter = None
        self._tools = None
        self._tool_names = (
            [sanitize_tool_name(name) for name in tool_names] if tool_names else None
        )

        if not MCP_AVAILABLE:
            import click

            if click.confirm(
                "You are missing the 'mcp' package. Would you like to install it?"
            ):
                import subprocess

                try:
                    subprocess.run(["uv", "add", "mcp crewai-tools'[mcp]'"], check=True)  # noqa: S607

                except subprocess.CalledProcessError as e:
                    raise ImportError("Failed to install mcp package") from e
            else:
                raise ImportError(
                    "`mcp` package not found, please run `uv add crewai-tools[mcp]`"
                )

        try:
            self._serverparams = serverparams
            self._adapter = MCPAdapt(
                self._serverparams, CrewAIToolAdapter(), connect_timeout
            )
            self.start()

        except Exception as e:
            if self._adapter is not None:
                try:
                    self.stop()
                except Exception as stop_e:
                    logger.error(f"Error during stop cleanup: {stop_e}")
            raise RuntimeError(f"Failed to initialize MCP Adapter: {e}") from e

    def start(self) -> None:
        """Start the MCP server and initialize the tools."""
        self._tools = self._adapter.__enter__()  # type: ignore[union-attr]

    def stop(self) -> None:
        """Stop the MCP server."""
        self._adapter.__exit__(None, None, None)  # type: ignore[union-attr]

    @property
    def tools(self) -> ToolCollection[BaseTool]:
        """The CrewAI tools available from the MCP server.

        Raises:
            ValueError: If the MCP server is not started.

        Returns:
            The CrewAI tools available from the MCP server.
        """
        if self._tools is None:
            raise ValueError(
                "MCP server not started, run `mcp_server.start()` first before accessing `tools`"
            )

        tools_collection = ToolCollection(self._tools)
        if self._tool_names:
            return tools_collection.filter_by_names(self._tool_names)
        return tools_collection

    def __enter__(self) -> ToolCollection[BaseTool]:
        """Enter the context manager.

        Note that `__init__()` already starts the MCP server,
        so tools should already be available.
        """
        return self.tools

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: Any,
    ) -> None:
        """Exit the context manager."""
        self._adapter.__exit__(exc_type, exc_value, traceback)  # type: ignore[union-attr]