Skip to content

Mcp

McpToolSpec #

Bases: BaseToolSpec

MCPToolSpec will get the tools from MCP Client (only need to implement ClientSession) and convert them to LlamaIndex's FunctionTool objects.

Parameters:

Name Type Description Default
client ClientSession

An MCP client instance implementing ClientSession, and it should support the following methods in ClientSession: - list_tools: List all tools. - call_tool: Call a tool.

 required
allowed_tools Optional[List[str]]

If set, only return tools with the specified names.

 None
Source code in llama-index-integrations/tools/llama-index-tools-mcp/llama_index/tools/mcp/base.py 
 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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class McpToolSpec(BaseToolSpec):
    """
    MCPToolSpec will get the tools from MCP Client (only need to implement ClientSession) and convert them to LlamaIndex's FunctionTool objects.

    Args:
        client: An MCP client instance implementing ClientSession, and it should support the following methods in ClientSession:
            - list_tools: List all tools.
            - call_tool: Call a tool.
        allowed_tools: If set, only return tools with the specified names.
    """

    def __init__(
        self,
        client: ClientSession,
        allowed_tools: Optional[List[str]] = None,
    ) -> None:
        self.client = client
        self.allowed_tools = allowed_tools if allowed_tools is not None else []

    async def fetch_tools(self) -> List[Any]:
        """
        An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

        Returns:
            A list of tools, each tool object needs to contain name, description, inputSchema properties.
        """
        response = await self.client.list_tools()
        tools = response.tools if hasattr(response, "tools") else []
        if self.allowed_tools:
            tools = [tool for tool in tools if tool.name in self.allowed_tools]
        return tools

    def _create_tool_fn(self, tool_name: str) -> Callable:
        """
        Create a tool call function for a specified MCP tool name. The function internally wraps the call_tool call to the MCP Client.
        """

        async def async_tool_fn(**kwargs):
            return await self.client.call_tool(tool_name, kwargs)

        return async_tool_fn

    async def to_tool_list_async(self) -> List[FunctionTool]:
        """
        Asynchronous method to convert MCP tools to FunctionTool objects.

        Returns:
            A list of FunctionTool objects.
        """
        tools_list = await self.fetch_tools()
        function_tool_list: List[FunctionTool] = []
        for tool in tools_list:
            fn = self._create_tool_fn(tool.name)
            # Create a Pydantic model based on the tool inputSchema
            model_schema = create_model_from_json_schema(
                tool.inputSchema, model_name=f"{tool.name}_Schema"
            )
            metadata = ToolMetadata(
                name=tool.name,
                description=tool.description,
                fn_schema=model_schema,
            )
            function_tool = FunctionTool.from_defaults(fn=fn, tool_metadata=metadata)
            function_tool_list.append(function_tool)
        return function_tool_list

    def to_tool_list(self) -> List[FunctionTool]:
        """
        Synchronous interface: Convert MCP Client tools to FunctionTool objects.
        Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

        Returns:
            A list of FunctionTool objects.
        """
        return patch_sync(self.to_tool_list_async)()

fetch_tools async #

fetch_tools() -> List[Any]

An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

Returns:

Type Description
List[Any]

A list of tools, each tool object needs to contain name, description, inputSchema properties.
Source code in llama-index-integrations/tools/llama-index-tools-mcp/llama_index/tools/mcp/base.py 
76
77
78
79
80
81
82
83
84
85
86
87
async def fetch_tools(self) -> List[Any]:
    """
    An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

    Returns:
        A list of tools, each tool object needs to contain name, description, inputSchema properties.
    """
    response = await self.client.list_tools()
    tools = response.tools if hasattr(response, "tools") else []
    if self.allowed_tools:
        tools = [tool for tool in tools if tool.name in self.allowed_tools]
    return tools

to_tool_list_async async #

to_tool_list_async() -> List[FunctionTool]

Asynchronous method to convert MCP tools to FunctionTool objects.

Returns:

Type Description
List[FunctionTool]

A list of FunctionTool objects.
Source code in llama-index-integrations/tools/llama-index-tools-mcp/llama_index/tools/mcp/base.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def to_tool_list_async(self) -> List[FunctionTool]:
    """
    Asynchronous method to convert MCP tools to FunctionTool objects.

    Returns:
        A list of FunctionTool objects.
    """
    tools_list = await self.fetch_tools()
    function_tool_list: List[FunctionTool] = []
    for tool in tools_list:
        fn = self._create_tool_fn(tool.name)
        # Create a Pydantic model based on the tool inputSchema
        model_schema = create_model_from_json_schema(
            tool.inputSchema, model_name=f"{tool.name}_Schema"
        )
        metadata = ToolMetadata(
            name=tool.name,
            description=tool.description,
            fn_schema=model_schema,
        )
        function_tool = FunctionTool.from_defaults(fn=fn, tool_metadata=metadata)
        function_tool_list.append(function_tool)
    return function_tool_list

to_tool_list #

to_tool_list() -> List[FunctionTool]

Synchronous interface: Convert MCP Client tools to FunctionTool objects. Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

Returns:

Type Description
List[FunctionTool]

A list of FunctionTool objects.
Source code in llama-index-integrations/tools/llama-index-tools-mcp/llama_index/tools/mcp/base.py 
123
124
125
126
127
128
129
130
131
def to_tool_list(self) -> List[FunctionTool]:
    """
    Synchronous interface: Convert MCP Client tools to FunctionTool objects.
    Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

    Returns:
        A list of FunctionTool objects.
    """
    return patch_sync(self.to_tool_list_async)()