Method call_tool

helpers/mcp_handler.py:1165–1178

Call a tool with the given input data

(
        self, tool_name: str, input_data: Dict[str, Any]
    )

        return MCPTool(agent=agent, name=tool_name, method=None, args={}, message="", loop_data=None)

    async def call_tool(
        self, tool_name: str, input_data: Dict[str, Any]
    ) -> CallToolResult:
        """Call a tool with the given input data"""
        server_name_part, tool_name_part = _split_qualified_tool_name(tool_name)
        matched_server = None
        with self.__lock:
            for server in self.servers:
                if server.name == server_name_part and server.has_tool(tool_name_part):
                    matched_server = server
                    break
        if matched_server is None:
            raise ValueError(f"Tool {tool_name} not found")
        return await matched_server.call_tool(tool_name_part, input_data)


T = TypeVar("T")