MCPToolSpec 将从 MCP Client 获取工具（只需实现 ClientSession）并将其转换为 LlamaIndex 的 FunctionTool 对象。
参数
名称
| 类型 |
描述 |
默认值 |
client |
ClientSession
|
实现 ClientSession 的 MCP 客户端实例，它应支持 ClientSession 中的以下方法： - list_tools: 列出所有工具。 - call_tool: 调用工具。
|
|
allowed_tools
|
可选[列表[str]]
|
如果设置，则仅返回指定名称的工具。
|
源代码位于 llama-index-integrations/tools/llama-index-tools-mcp/llama_index/tools/mcp/base.py
|
无
|
fetch_tools 异步
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
132
133
134
135
136 | 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() -> List[Any]
返回
列表[Any]
| 描述 |
默认值 |
一个工具列表，每个工具对象需要包含 name、description、inputSchema 属性。
|
|
fetch_tools 异步
78
79
80
81
82
83
84
85
86
87
88
89
90 | 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
|
列表[FunctionTool]
列表[Any]
| 描述 |
默认值 |
FunctionTool 对象的列表。
|
|
fetch_tools 异步
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125 | 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
|
回到顶部
列表[Any]
| 描述 |
默认值 |
FunctionTool 对象的列表。
|
|
fetch_tools 异步
127
128
129
130
131
132
133
134
135
136 | 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)()
|
