Initial dynamic_list support (#764)

# Description

Implementing `dynamic_list` contract to unlock the scenario that any
tool should be able to define the listing experience and the
corresponding request to enable the dynamic listing feature. See more in
#383

In this PR, we did following things:
- implement `gen_dynamic_list` in `tools_manager`.
- append ws triple to function input params, as user may call azure
control plane api in the custom function to list Azure resources.
- add an example tool with dynamic list input to `my_tool_package`.

# All Promptflow Contribution checklist:
- [X] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [X] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [X] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [X] Title of the pull request is clear and informative.
- [X] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [X] Pull request includes test coverage for the included changes.

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.