remove toolset.get_meta_tools() and update to new API

Author: shashi-stackone

examples/meta_tools_example.py

@@ -9,13 +9,12 @@
    (tool_search + tool_execute). The LLM discovers and runs tools on-demand,
    keeping token usage constant regardless of catalog size.
 
-This example demonstrates approach 2 with OpenAI and LangChain clients.
+This example demonstrates approach 2 with a Gemini client (OpenAI-compatible).
 
 Prerequisites:
     - STACKONE_API_KEY environment variable
     - STACKONE_ACCOUNT_ID environment variable
     - GOOGLE_API_KEY environment variable (for Gemini)
-    - OPENAI_API_KEY environment variable (optional, for LangChain example)
 
 Run with:
     uv run python examples/meta_tools_example.py
@@ -68,8 +67,7 @@ def example_gemini() -> None:
         execute={"account_ids": [account_id]} if account_id else None,
     )
 
-    # 2. Get meta tools in OpenAI format
-    meta_tools = toolset.get_meta_tools()
+    # 2. Get tools in OpenAI format
     openai_tools = toolset.openai(mode="search_and_execute")
 
     # 3. Create Gemini client (OpenAI-compatible) and run agent loop
@@ -100,8 +98,7 @@ def example_gemini() -> None:
         messages.append(choice.message.model_dump(exclude_none=True))
         for tool_call in choice.message.tool_calls:
             print(f"  -> {tool_call.function.name}({tool_call.function.arguments})")
-            tool = meta_tools.get_tool(tool_call.function.name)
-            result = tool.execute(tool_call.function.arguments) if tool else {"error": "Unknown tool"}
+            result = toolset.execute(tool_call.function.name, tool_call.function.arguments)
             messages.append(
                 {
                     "role": "tool",
@@ -113,64 +110,6 @@ def example_gemini() -> None:
     print()
 
 
-def example_langchain() -> None:
-    """Complete LangChain integration with meta tools.
-
-    Shows: init toolset -> bind tools to ChatOpenAI -> agent loop -> final answer.
-    """
-    print("=" * 60)
-    print("Example 2: LangChain client with meta tools")
-    print("=" * 60)
-    print()
-
-    try:
-        from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
-        from langchain_google_genai import ChatGoogleGenerativeAI
-    except ImportError:
-        print("Skipped: pip install langchain-google-genai")
-        print()
-        return
-
-    if not os.getenv("GOOGLE_API_KEY"):
-        print("Skipped: Set GOOGLE_API_KEY to run this example.")
-        print()
-        return
-
-    # 1. Init toolset
-    account_id = os.getenv("STACKONE_ACCOUNT_ID")
-    toolset = StackOneToolSet(
-        account_id=account_id,
-        search={"method": "semantic", "top_k": 3},
-        execute={"account_ids": [account_id]} if account_id else None,
-    )
-
-    # 2. Get meta tools in LangChain format and bind to model
-    meta_tools = toolset.get_meta_tools()
-    langchain_tools = meta_tools.to_langchain()
-    model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview").bind_tools(langchain_tools)
-
-    # 3. Run agent loop
-    messages = [HumanMessage(content="List my upcoming Calendly events for the next week.")]
-
-    for _step in range(10):
-        response: AIMessage = model.invoke(messages)
-
-        # 4. If no tool calls, print final answer and stop
-        if not response.tool_calls:
-            print(f"Answer: {response.content}")
-            break
-
-        # 5. Execute tool calls and feed results back
-        messages.append(response)
-        for tool_call in response.tool_calls:
-            print(f"  -> {tool_call['name']}({json.dumps(tool_call['args'])})")
-            tool = meta_tools.get_tool(tool_call["name"])
-            result = tool.execute(tool_call["args"]) if tool else {"error": "Unknown tool"}
-            messages.append(ToolMessage(content=json.dumps(result), tool_call_id=tool_call["id"]))
-
-    print()
-
-
 def main() -> None:
     """Run all meta tools examples."""
     api_key = os.getenv("STACKONE_API_KEY")
@@ -179,7 +118,6 @@ def main() -> None:
         return
 
     example_gemini()
-    example_langchain()
 
 
 if __name__ == "__main__":

stackone_ai/toolset.py

@@ -367,6 +367,7 @@ def __init__(
         self._semantic_client: SemanticSearchClient | None = None
         self._search_config: SearchConfig | None = search
         self._execute_config: ExecuteToolsConfig | None = execute
+        self._meta_tools_cache: Tools | None = None
 
     def set_accounts(self, account_ids: list[str]) -> StackOneToolSet:
         """Set account IDs for filtering tools
@@ -502,6 +503,36 @@ def openai(
 
         return self.fetch_tools(account_ids=effective_account_ids).to_openai()
 
+
+    def execute(
+        self,
+        tool_name: str,
+        arguments: str | dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Execute a tool by name.
+
+        Use with ``openai(mode="search_and_execute")`` in manual agent loops —
+        pass the tool name and arguments from the LLM's tool call directly.
+
+        Meta tools are cached after the first call.
+
+        Args:
+            tool_name: The tool name from the LLM's tool call
+                (e.g. ``"tool_search"`` or ``"tool_execute"``).
+            arguments: The arguments from the LLM's tool call,
+                as a JSON string or dict.
+
+        Returns:
+            Tool execution result as a dict.
+        """
+        if self._meta_tools_cache is None:
+            self._meta_tools_cache = self.get_meta_tools()
+
+        tool = self._meta_tools_cache.get_tool(tool_name)
+        if tool is None:
+            return {"error": f'Tool "{tool_name}" not found.'}
+        return tool.execute(arguments)
+
     @property
     def semantic_client(self) -> SemanticSearchClient:
         """Lazy initialization of semantic search client.

tests/test_meta_tools.py

@@ -431,3 +431,57 @@ def test_openai_search_and_execute_with_execute_config(self):
             toolset.openai(mode="search_and_execute")
 
         mock_get.assert_called_once_with(account_ids=["acc-1"])
+
+
+class TestToolSetExecuteMethod:
+    """Tests for StackOneToolSet.execute() convenience method."""
+
+    def test_execute_delegates_to_meta_tool(self):
+        toolset = StackOneToolSet(api_key="test-key", search={"method": "auto"})
+        mock_tool = MagicMock()
+        mock_tool.execute.return_value = {"result": "ok"}
+        mock_meta = MagicMock()
+        mock_meta.get_tool.return_value = mock_tool
+
+        with patch.object(toolset, "get_meta_tools", return_value=mock_meta):
+            result = toolset.execute("tool_search", {"query": "employees"})
+
+        assert result == {"result": "ok"}
+        mock_meta.get_tool.assert_called_once_with("tool_search")
+        mock_tool.execute.assert_called_once_with({"query": "employees"})
+
+    def test_execute_caches_meta_tools(self):
+        toolset = StackOneToolSet(api_key="test-key", search={"method": "auto"})
+        mock_tool = MagicMock()
+        mock_tool.execute.return_value = {"ok": True}
+        mock_meta = MagicMock()
+        mock_meta.get_tool.return_value = mock_tool
+
+        with patch.object(toolset, "get_meta_tools", return_value=mock_meta) as mock_get:
+            toolset.execute("tool_search", {"query": "a"})
+            toolset.execute("tool_execute", {"tool_name": "b"})
+
+        mock_get.assert_called_once()
+
+    def test_execute_returns_error_for_unknown_tool(self):
+        toolset = StackOneToolSet(api_key="test-key", search={"method": "auto"})
+        mock_meta = MagicMock()
+        mock_meta.get_tool.return_value = None
+
+        with patch.object(toolset, "get_meta_tools", return_value=mock_meta):
+            result = toolset.execute("nonexistent", {})
+
+        assert "error" in result
+
+    def test_execute_accepts_string_arguments(self):
+        toolset = StackOneToolSet(api_key="test-key", search={"method": "auto"})
+        mock_tool = MagicMock()
+        mock_tool.execute.return_value = {"ok": True}
+        mock_meta = MagicMock()
+        mock_meta.get_tool.return_value = mock_tool
+
+        with patch.object(toolset, "get_meta_tools", return_value=mock_meta):
+            result = toolset.execute("tool_search", '{"query": "test"}')
+
+        assert result == {"ok": True}
+        mock_tool.execute.assert_called_once_with('{"query": "test"}')