feat: Create helper methods that return a future of the asynchronous tasks scheduled on an internal background loop/thread

anubhav756 · anubhav756 · commit 2a4b36db4d5c · 2025-05-17T01:19:16.000+05:30
diff --git a/packages/toolbox-core/src/toolbox_core/sync_client.py b/packages/toolbox-core/src/toolbox_core/sync_client.py
@@ -15,6 +15,7 @@
 
 import asyncio
 from asyncio import AbstractEventLoop
+from concurrent.futures import Future
 from threading import Thread
 from typing import Any, Callable, Coroutine, Mapping, Optional, Union
 
@@ -86,6 +87,103 @@ def close(self):
         coro = self.__async_client.close()
         asyncio.run_coroutine_threadsafe(coro, self.__loop).result()
 
+    def _load_tool_future(
+        self,
+        name: str,
+        auth_token_getters: dict[str, Callable[[], str]] = {},
+        bound_params: Mapping[str, Union[Callable[[], Any], Any]] = {},
+    ) -> Future[ToolboxSyncTool]:
+        """
+        Asynchronously initiates the loading of a specific tool from the Toolbox service.
+
+        This method schedules the tool loading operation on a background event loop
+        and immediately returns a `concurrent.futures.Future`. This allows other
+        operations to proceed while the tool is being loaded. To get the actual
+        `ToolboxSyncTool` instance, call `.result()` on the returned future, which
+        will block until the tool is available or an error occurs.
+
+        Args:
+            name: The unique name or identifier of the tool to load.
+            auth_token_getters: A mapping of authentication service names to
+                callables that return the corresponding authentication token.
+            bound_params: A mapping of parameter names to bind to specific values or
+                callables that are called to produce values as needed.
+
+        Returns:
+            A `concurrent.futures.Future` that, upon successful completion, will
+            yield a `ToolboxSyncTool` instance representing the loaded tool.
+
+        Raises:
+            ValueError: If the background event loop or thread (required for asynchronous
+                        operations) is not running or properly initialized.
+
+        """
+
+        async def async_worker() -> ToolboxSyncTool:
+            if not self.__loop or not self.__thread:
+                raise ValueError("Background loop or thread cannot be None.")
+            async_tool = await self.__async_client.load_tool(
+                name, auth_token_getters, bound_params
+            )
+            return ToolboxSyncTool(async_tool, self.__loop, self.__thread)
+
+        if not self.__loop or not self.__thread:
+            raise ValueError("Background loop or thread cannot be None.")
+        return asyncio.run_coroutine_threadsafe(async_worker(), self.__loop)
+
+    def _load_toolset_future(
+        self,
+        name: Optional[str] = None,
+        auth_token_getters: dict[str, Callable[[], str]] = {},
+        bound_params: Mapping[str, Union[Callable[[], Any], Any]] = {},
+        strict: bool = False,
+    ) -> Future[list[ToolboxSyncTool]]:
+        """
+        Asynchronously initiates loading of all tools within a specified toolset.
+
+        This method schedules the toolset loading operation on a background event
+        loop and returns a `concurrent.futures.Future` without blocking.
+        The future's result will be a list of `ToolboxSyncTool` instances.
+        Call `.result()` on the returned future to wait for completion and get
+        the list of tools.
+
+        Args:
+            name: Name of the toolset to load tools.
+            auth_token_getters: A mapping of authentication service names to
+                callables that return the corresponding authentication token.
+            bound_params: A mapping of parameter names to bind to specific values or
+                callables that are called to produce values as needed.
+            strict: If True, raises an error if *any* loaded tool instance fails
+                to utilize at least one provided parameter or auth token (if any
+                provided). If False (default), raises an error only if a
+                user-provided parameter or auth token cannot be applied to *any*
+                loaded tool across the set.
+
+        Returns:
+            A `concurrent.futures.Future` that, upon successful completion, will
+            yield a list of `ToolboxSyncTool` instances.
+
+        Raises:
+            ValueError: If the background event loop or thread is not running,
+                        or if validation fails based on the `strict` flag during
+                        the underlying asynchronous loading process.
+        """
+
+        async def async_worker() -> list[ToolboxSyncTool]:
+            if not self.__loop or not self.__thread:
+                raise ValueError("Background loop or thread cannot be None.")
+            async_tools = await self.__async_client.load_toolset(
+                name, auth_token_getters, bound_params, strict
+            )
+            return [
+                ToolboxSyncTool(async_tool, self.__loop, self.__thread)
+                for async_tool in async_tools
+            ]
+
+        if not self.__loop or not self.__thread:
+            raise ValueError("Background loop or thread cannot be None.")
+        return asyncio.run_coroutine_threadsafe(async_worker(), self.__loop)
+
     def load_tool(
         self,
         name: str,
@@ -111,13 +209,7 @@ def load_tool(
                 for execution. The specific arguments and behavior of the callable
                 depend on the tool itself.
         """
-        coro = self.__async_client.load_tool(name, auth_token_getters, bound_params)
-
-        if not self.__loop or not self.__thread:
-            raise ValueError("Background loop or thread cannot be None.")
-
-        async_tool = asyncio.run_coroutine_threadsafe(coro, self.__loop).result()
-        return ToolboxSyncTool(async_tool, self.__loop, self.__thread)
+        return self._load_tool_future(name, auth_token_getters, bound_params).result()
 
     def load_toolset(
         self,
@@ -148,18 +240,9 @@ def load_toolset(
         Raises:
             ValueError: If validation fails based on the `strict` flag.
         """
-        coro = self.__async_client.load_toolset(
+        return self._load_toolset_future(
             name, auth_token_getters, bound_params, strict
-        )
-
-        if not self.__loop or not self.__thread:
-            raise ValueError("Background loop or thread cannot be None.")
-
-        async_tools = asyncio.run_coroutine_threadsafe(coro, self.__loop).result()
-        return [
-            ToolboxSyncTool(async_tool, self.__loop, self.__thread)
-            for async_tool in async_tools
-        ]
+        ).result()
 
     def add_headers(
         self, headers: Mapping[str, Union[Callable, Coroutine, str]]
diff --git a/packages/toolbox-core/src/toolbox_core/sync_tool.py b/packages/toolbox-core/src/toolbox_core/sync_tool.py
@@ -15,6 +15,7 @@
 
 import asyncio
 from asyncio import AbstractEventLoop
+from concurrent.futures import Future
 from inspect import Signature
 from threading import Thread
 from typing import Any, Callable, Coroutine, Mapping, Sequence, Union
@@ -129,6 +130,32 @@ def _auth_service_token_getters(self) -> Mapping[str, Callable[[], str]]:
     def _client_headers(self) -> Mapping[str, Union[Callable, Coroutine, str]]:
         return self.__async_tool._client_headers
 
+    def _call_future(self, *args: Any, **kwargs: Any) -> Future[str]:
+        """
+        Asynchronously calls the remote tool with the provided arguments.
+
+        This method schedules the actual remote tool invocation on a background
+        event loop and immediately returns a `concurrent.futures.Future`.
+        This allows the caller to perform other operations while the remote
+        tool execution is in progress.
+
+        To obtain the result of the tool's execution, call `.result()` on
+        the returned Future object. This will block until the remote tool
+        responds or an error occurs.
+
+        Args:
+            *args: Positional arguments for the tool.
+            **kwargs: Keyword arguments for the tool.
+
+        Returns:
+            A `concurrent.futures.Future` that will eventually resolve to the
+            string result returned by the remote tool's execution. If the
+            remote tool call fails, the future will resolve to an exception.
+
+        """
+        coro = self.__async_tool(*args, **kwargs)
+        return asyncio.run_coroutine_threadsafe(coro, self.__loop)
+
     def __call__(self, *args: Any, **kwargs: Any) -> str:
         """
         Synchronously calls the remote tool with the provided arguments.
@@ -142,9 +169,9 @@ def __call__(self, *args: Any, **kwargs: Any) -> str:
 
         Returns:
             The string result returned by the remote tool execution.
+
         """
-        coro = self.__async_tool(*args, **kwargs)
-        return asyncio.run_coroutine_threadsafe(coro, self.__loop).result()
+        return self._call_future(*args, **kwargs).result()
 
     def add_auth_token_getters(
         self,
