feat: add support for bound parameters

Author: kurtisvg

packages/toolbox-core/src/toolbox_core/client.py

@@ -59,18 +59,22 @@ def __parse_tool(
         name: str,
         schema: ToolSchema,
         auth_token_getters: dict[str, Callable[[], str]],
+        all_bound_params: dict[str, Callable[[], str]],
     ) -> ToolboxTool:
         """Internal helper to create a callable tool from its schema."""
-        # sort into authenticated and reg params
+        # sort into reg, authn, and bound params
         params = []
         authn_params: dict[str, list[str]] = {}
+        bound_params: dict[str, Callable[[], str]] = {}
         auth_sources: set[str] = set()
         for p in schema.parameters:
-            if not p.authSources:
-                params.append(p)
-            else:
+            if p.authSources:  # authn parameter
                 authn_params[p.name] = p.authSources
                 auth_sources.update(p.authSources)
+            elif p.name in all_bound_params:  # bound parameter
+                bound_params[p.name] = all_bound_params[p.name]
+            else:  # regular parameter
+                params.append(p)
 
         authn_params = identify_required_authn_params(
             authn_params, auth_token_getters.keys()
@@ -85,6 +89,7 @@ def __parse_tool(
             # create a read-only values for the maps to prevent mutation
             required_authn_params=types.MappingProxyType(authn_params),
             auth_service_token_getters=types.MappingProxyType(auth_token_getters),
+            bound_params=types.MappingProxyType(bound_params),
         )
         return tool
 
@@ -124,6 +129,7 @@ async def load_tool(
         self,
         name: str,
         auth_token_getters: dict[str, Callable[[], str]] = {},
+        bound_params: dict[str, Callable[[], str]] = {},
     ) -> ToolboxTool:
         """
         Asynchronously loads a tool from the server.
@@ -154,14 +160,17 @@ async def load_tool(
         if name not in manifest.tools:
             # TODO: Better exception
             raise Exception(f"Tool '{name}' not found!")
-        tool = self.__parse_tool(name, manifest.tools[name], auth_token_getters)
+        tool = self.__parse_tool(
+            name, manifest.tools[name], auth_token_getters, bound_params
+        )
 
         return tool
 
     async def load_toolset(
         self,
         name: str,
         auth_token_getters: dict[str, Callable[[], str]] = {},
+        bound_params: dict[str, Callable[[], str]] = {},
     ) -> list[ToolboxTool]:
         """
         Asynchronously fetches a toolset and loads all tools defined within it.
@@ -172,6 +181,7 @@ async def load_toolset(
                 callables that return the corresponding authentication token.
 
 
+
         Returns:
             list[ToolboxTool]: A list of callables, one for each tool defined
             in the toolset.
@@ -184,7 +194,7 @@ async def load_toolset(
 
         # parse each tools name and schema into a list of ToolboxTools
         tools = [
-            self.__parse_tool(n, s, auth_token_getters)
+            self.__parse_tool(n, s, auth_token_getters, bound_params)
             for n, s in manifest.tools.items()
         ]
         return tools

packages/toolbox-core/src/toolbox_core/tool.py

@@ -13,10 +13,20 @@
 # limitations under the License.
 
 
+import asyncio
 import types
 from collections import defaultdict
 from inspect import Parameter, Signature
-from typing import Any, Callable, DefaultDict, Iterable, Mapping, Optional, Sequence
+from typing import (
+    Any,
+    Callable,
+    DefaultDict,
+    Iterable,
+    Mapping,
+    Optional,
+    Sequence,
+    Union,
+)
 
 from aiohttp import ClientSession
 from pytest import Session
@@ -44,6 +54,7 @@ def __init__(
         params: Sequence[Parameter],
         required_authn_params: Mapping[str, list[str]],
         auth_service_token_getters: Mapping[str, Callable[[], str]],
+        bound_params: Mapping[str, Union[Callable[[], Any], Any]] = {},
     ):
         """
         Initializes a callable that will trigger the tool invocation through the
@@ -81,6 +92,8 @@ def __init__(
         self.__required_authn_params = required_authn_params
         # map of authService -> token_getter
         self.__auth_service_token_getters = auth_service_token_getters
+        # map of parameter name to value or Callable
+        self.__bound_parameters = bound_params
 
     def __copy(
         self,
@@ -91,6 +104,7 @@ def __copy(
         params: Optional[list[Parameter]] = None,
         required_authn_params: Optional[Mapping[str, list[str]]] = None,
         auth_service_token_getters: Optional[Mapping[str, Callable[[], str]]] = None,
+        bound_params: Optional[Mapping[str, Union[Callable[[], Any], Any]]] = None,
     ) -> "ToolboxTool":
         """
         Creates a copy of the ToolboxTool, overriding specific fields.
@@ -121,6 +135,7 @@ def __copy(
             auth_service_token_getters=check(
                 auth_service_token_getters, self.__auth_service_token_getters
             ),
+            bound_params=check(bound_params, self.__bound_parameters),
         )
 
     async def __call__(self, *args: Any, **kwargs: Any) -> str:
@@ -153,6 +168,14 @@ async def __call__(self, *args: Any, **kwargs: Any) -> str:
         all_args.apply_defaults()  # Include default values if not provided
         payload = all_args.arguments
 
+        # apply bounded parameters
+        for param, value in self.__bound_parameters.items():
+            if asyncio.iscoroutinefunction(value):
+                value = await value()
+            elif callable(value):
+                value = value()
+            payload[param] = value
+
         # create headers for auth services
         headers = {}
         for auth_service, token_getter in self.__auth_service_token_getters.items():
@@ -211,13 +234,41 @@ def add_auth_token_getters(
             required_authn_params=new_req_authn_params,
         )
 
+    def bind_parameters(
+        self, bound_params: Mapping[str, Callable[[], str]]
+    ) -> "ToolboxTool":
+        """
+        Binds parameters to values or callables that produce values.
+
+         Args:
+             bound_params: A mapping of parameter names to values or callables that
+                 produce values.
+
+         Returns:
+             A new ToolboxTool instance with the specified parameters bound.
+        """
+        all_params = set(p.name for p in self.__params)
+        for name in bound_params.keys():
+            if name not in all_params:
+                raise Exception(f"unable to bind parameters: no parameter named {name}")
+
+        new_params = []
+        for p in self.__params:
+            if p.name not in bound_params:
+                new_params.append(p)
+
+        return self.__copy(
+            params=new_params,
+            bound_params=bound_params,
+        )
+
 
 def identify_required_authn_params(
     req_authn_params: Mapping[str, list[str]], auth_service_names: Iterable[str]
 ) -> dict[str, list[str]]:
     """
-    Identifies authentication parameters that are still required; or not covered by
-        the provided `auth_service_names`.
+    Identifies authentication parameters that are still required; because they
+        not covered by the provided `auth_service_names`.
 
     Args:
         req_authn_params: A mapping of parameter names to sets of required
@@ -226,8 +277,8 @@ def identify_required_authn_params(
             token getters are available.
 
     Returns:
-        A new dictionary representing the subset of required authentication
-        parameters that are not covered by the provided `auth_service_names`.
+        A new dictionary representing the subset of required authentication parameters
+        that are not covered by the provided `auth_services`.
     """
     required_params = {}  # params that are still required with provided auth_services
     for param, services in req_authn_params.items():