Initial dynamic_list support

examples/tools/tool-package-quickstart/my_tool_package/tools/tool_with_dynamic_list_input.py

@@ -0,0 +1,36 @@
+from promptflow import tool
+from typing import List, Union, Dict
+
+
+def my_list_func(prefix: str = "", size: int = 10, **kwargs) -> List[Dict[str, Union[str, int, float, list, Dict]]]:
+    """This is a dummy function to generate a list of items.
+
+    :param prefix: prefix to add to each item.
+    :param size: number of items to generate.
+    :param kwargs: other parameters.
+    :return: a list of items. Each item is a dict with the following keys:
+        - value: for backend use. Required.
+        - display_value: for UI display. Optional.
+        - hyperlink: external link. Optional.
+        - description: information icon tip. Optional.
+    """
+    import random
+
+    words = ["apple", "banana", "cherry", "date", "elderberry", "fig", "grape", "honeydew", "kiwi", "lemon"]
+    result = []
+    for i in range(size):
+        random_word = f"{random.choice(words)}{i}"
+        cur_item = {
+            "value": random_word,
+            "display_value": f"{prefix}_{random_word}",
+            "hyperlink": f'https://www.google.com/search?q={random_word}',
+            "description": f"this is {i} item",
+        }
+        result.append(cur_item)
+
+    return result
+
+
+@tool
+def my_tool(input_text: list, input_prefix: str) -> str:
+    return f"Hello {input_prefix} {','.join(input_text)}"

examples/tools/tool-package-quickstart/my_tool_package/yamls/tool_with_dynamic_list_input.yaml

@@ -0,0 +1,35 @@
+my_tool_package.tools.tool_with_dynamic_list_input.my_tool:
+  function: my_tool
+  inputs:
+    input_text:
+      type:
+      - list
+      dynamic_list:
+        # UX send dynamic_list content to backend.
+        # specifies the function to generate dynamic list. format: <module>.<func>
+        func_path: my_tool_package.tools.tool_with_dynamic_list_input.my_list_func
+        func_kwargs: 
+        - name: prefix  # Argument name to be passed to the function
+          type: 
+          - string
+          # if optional is not specified, default to false.
+          # this is for UX pre-validaton. If optional is false, but no input. UX can throw error in advanced.
+          optional: true
+          reference: ${inputs.input_prefix}  # Dynamic reference to another input parameter
+        - name: size  # Another argument name to be passed to the function
+          type: 
+          - int
+          optional: true
+          default: 10
+      # enum and dynamic list may need below setting, default false.
+      # allow user to enter input `index_path` value manually.
+      allow_manual_entry: true
+      is_multi_select: true
+    # used to filter 
+    input_prefix:
+      type:
+      - string
+  module: my_tool_package.tools.tool_with_dynamic_list_input
+  name: My Tool with Dynamic List Input
+  description: This is my tool with dynamic list input
+  type: python

examples/tools/tool-package-quickstart/tests/test_tool_with_dynamic_input.py

@@ -0,0 +1,12 @@
+from my_tool_package.tools.tool_with_dynamic_list_input import my_tool, my_list_func
+
+
+def test_my_tool():
+    result = my_tool(input_text=["apple", "banana"], input_prefix="My")
+    assert result == 'Hello My apple,banana'
+
+
+def test_my_list_func():
+    result = my_list_func(prefix="My")
+    assert len(result) == 10
+    assert "value" in result[0]

src/promptflow/promptflow/_core/tools_manager.py

@@ -10,7 +10,7 @@
 import types
 from functools import partial
 from pathlib import Path
-from typing import Callable, List, Mapping, Optional, Tuple, Union
+from typing import Callable, List, Mapping, Optional, Tuple, Union, Dict
 
 import pkg_resources
 import yaml
@@ -170,6 +170,50 @@ def gen_tool_by_source(name, source: ToolSource, tool_type: ToolType, working_di
             )
 
 
+def append_workspace_triple_to_func_input_params(func_sig_params, func_input_params_dict, ws_triple_dict):
+    '''Append workspace triple to func input params.
+
+    :param func_sig_params: function signature parameters, full params.
+    :param func_input_params_dict: user input param key-values for dynamic list function.
+    :param ws_triple_dict: workspace triple dict, including subscription_id, resource_group_name, workspace_name.
+    :return: combined func input params.
+    '''
+    # append workspace triple to func input params if any below condition are met:
+    # 1. func signature has kwargs param.
+    # 2. func signature has param named 'subscription_id','resource_group_name','workspace_name'.
+    has_kwargs_param = any([param.kind == inspect.Parameter.VAR_KEYWORD for _, param in func_sig_params.items()])
+    if has_kwargs_param is False:
+        # keep only params that are in func signature. Or run into error when calling func.
+        avail_ws_info_dict = {k: v for k, v in ws_triple_dict.items() if k in set(func_sig_params.keys())}
+    else:
+        avail_ws_info_dict = ws_triple_dict
+
+    # if ws triple key is in func input params, it means user has provided value for it,
+    # do not expect implicit override.
+    combined_func_input_params = dict(avail_ws_info_dict, **func_input_params_dict)
+    return combined_func_input_params
+
+
+def gen_dynamic_list(func_path: str, func_input_params_dict: Dict, ws_triple_dict: Dict[str, str] = None):
+    import importlib
+    import inspect
+
+    # TODO: validate func path.
+    module_name, func_name = func_path.rsplit('.', 1)
+    module = importlib.import_module(module_name)
+    func = getattr(module, func_name)
+    # get param names from func signature.
+    func_sig_params = inspect.signature(func).parameters
+    # TODO: validate if func input params are all in func signature params.
+    # TODO: add more tests to verify following appending logic.
+    combined_func_input_params = append_workspace_triple_to_func_input_params(
+        func_sig_params, func_input_params_dict, ws_triple_dict)
+    # TODO: error handling of func call.
+    result = func(**combined_func_input_params)
+    # TODO: validate response is of required format. Throw correct message if response is empty.
+    return result
+
+
 class BuiltinsManager:
     def __init__(self) -> None:
         pass

src/promptflow/promptflow/_internal/__init__.py

@@ -40,6 +40,7 @@
     register_apis,
     register_builtins,
     register_connections,
+    gen_dynamic_list,
 )
 from promptflow._core.tracer import Tracer
 from promptflow._sdk._constants import LOCAL_MGMT_DB_PATH

src/promptflow/promptflow/_sdk/_utils.py

@@ -32,6 +32,7 @@
 import promptflow
 from promptflow._constants import EXTENSION_UA
 from promptflow._core.tool_meta_generator import generate_tool_meta_dict_by_file
+from promptflow._core.tools_manager import gen_dynamic_list
 from promptflow._sdk._constants import (
     DAG_FILE_NAME,
     DEFAULT_ENCODING,
@@ -624,6 +625,28 @@ def _generate_tool_meta(
     return res
 
 
+def _gen_dynamic_list(function_config: Dict) -> List:
+    """Generate dynamic list for a tool input.
+
+    :param function_config: function config in tool meta. Should contain'func_path' and 'func_kwargs'.
+    :return: a list of tool input dynamic enums.
+    """
+    func_path = function_config.get("func_path", "")
+    func_kwargs = function_config.get("func_kwargs", {})
+    # May call azure control plane api in the custom function to list Azure resources.
+    # which may need Azure workspace triple.
+    # TODO: move this method to a common place.
+    from promptflow._cli._utils import get_workspace_triad_from_local
+
+    workspace_triad = get_workspace_triad_from_local()
+    if (workspace_triad.subscription_id and workspace_triad.resource_group_name
+            and workspace_triad.workspace_name):
+        return gen_dynamic_list(func_path, func_kwargs, workspace_triad._asdict())
+    # if no workspace triple available, just skip.
+    else:
+        return gen_dynamic_list(func_path, func_kwargs)
+
+
 def _generate_package_tools(keys: Optional[List[str]] = None) -> dict:
     import imp
 

src/promptflow/tests/executor/unittests/_core/test_tools_manager.py

@@ -199,3 +199,15 @@ def test_collect_package_tools_and_connections(self, install_custom_tool_pkg):
         content = templates["my_tool_package.tools.my_tool_with_custom_strong_type_connection.MyCustomConnection"]
         expected_template_str = textwrap.dedent(expected_template)
         assert expected_template_str in content
+
+    # TODO: enable this test after new my_tool_package is released
+    @pytest.mark.skip("Will enable this test after new my_tool_package is released")
+    def test_gen_dynamic_list(self):
+        from promptflow._sdk._utils import _gen_dynamic_list
+        func_path = "my_tool_package.tools.tool_with_dynamic_list_input.my_list_func"
+        func_kwargs = {"prefix": "My"}
+        result = _gen_dynamic_list({
+            "func_path": func_path, "func_kwargs": func_kwargs})
+        assert len(result) == 10
+
+    # TODO: add test for gen_dynamic_list with ws_triple.