Skip to content

CrewAI Adapter

This module implements the CrewAI adapter.

CrewAI tools support only sync functions for their tools.

Example Usage:

with MCPAdapt(StdioServerParameters(command="uv", args=["run", "src/echo.py"]), CrewAIAdapter()) as tools: print(tools)

CrewAIAdapter

Bases: ToolAdapter

Adapter for crewai.

Note that crewai support only sync tools so we write adapt for sync tools only.

Warning: if the mcp tool name is a python keyword, starts with digits or contains dashes, the tool name will be sanitized to become a valid python function name.

Source code in src/mcpadapt/crewai_adapter.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class CrewAIAdapter(ToolAdapter):
    """Adapter for `crewai`.

    Note that `crewai` support only sync tools so we write adapt for sync tools only.

    Warning: if the mcp tool name is a python keyword, starts with digits or contains
    dashes, the tool name will be sanitized to become a valid python function name.

    """

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

        Args:
            func: The function to adapt.
            mcp_tool: The MCP tool to adapt.

        Returns:
            A CrewAI tool.
        """
        ToolInput = create_model_from_json_schema(mcp_tool.inputSchema)

        class CrewAIMCPTool(BaseTool):
            name: str = mcp_tool.name
            description: str = mcp_tool.description or ""
            args_schema: Type[BaseModel] = ToolInput

            def _run(self, *args: Any, **kwargs: Any) -> Any:
                print("args", args)
                print("kwargs", kwargs)
                return func(kwargs).content[0].text  # type: ignore

            def _generate_description(self):
                args_schema = {
                    k: v
                    for k, v in jsonref.replace_refs(
                        self.args_schema.model_json_schema()
                    ).items()
                    if k != "$defs"
                }
                self.description = f"Tool Name: {self.name}\nTool Arguments: {args_schema}\nTool Description: {self.description}"

        return CrewAIMCPTool()

    async def async_adapt(
        self,
        afunc: Callable[[dict | None], Coroutine[Any, Any, mcp.types.CallToolResult]],
        mcp_tool: mcp.types.Tool,
    ) -> Any:
        raise NotImplementedError("async is not supported by the CrewAI framework.")

adapt 

adapt(
    func: Callable[[dict | None], CallToolResult],
    mcp_tool: Tool,
) -> BaseTool

Adapt a MCP tool to a CrewAI tool.

Parameters:

Name Type Description Default
func Callable[[dict | None], CallToolResult]

The function to adapt.

 required
mcp_tool Tool

The MCP tool to adapt.

 required

Returns:

Type Description
BaseTool

A CrewAI tool.
Source code in src/mcpadapt/crewai_adapter.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def adapt(
    self,
    func: Callable[[dict | None], mcp.types.CallToolResult],
    mcp_tool: mcp.types.Tool,
) -> BaseTool:
    """Adapt a MCP tool to a CrewAI tool.

    Args:
        func: The function to adapt.
        mcp_tool: The MCP tool to adapt.

    Returns:
        A CrewAI tool.
    """
    ToolInput = create_model_from_json_schema(mcp_tool.inputSchema)

    class CrewAIMCPTool(BaseTool):
        name: str = mcp_tool.name
        description: str = mcp_tool.description or ""
        args_schema: Type[BaseModel] = ToolInput

        def _run(self, *args: Any, **kwargs: Any) -> Any:
            print("args", args)
            print("kwargs", kwargs)
            return func(kwargs).content[0].text  # type: ignore

        def _generate_description(self):
            args_schema = {
                k: v
                for k, v in jsonref.replace_refs(
                    self.args_schema.model_json_schema()
                ).items()
                if k != "$defs"
            }
            self.description = f"Tool Name: {self.name}\nTool Arguments: {args_schema}\nTool Description: {self.description}"

    return CrewAIMCPTool()