fix: tools

Author: Matt Carey

examples/available_tools.py

@@ -7,6 +7,7 @@
 """
 
 
+# TODO: Add examples
 def get_available_tools():
     print("Getting available tools")
 

examples/crewai_integration.py

@@ -8,9 +8,41 @@
 ```
 """
 
+from crewai import Agent, Crew, Task
+from stackone_ai import StackOneToolSet
+
+account_id = "45072196112816593343"
+employee_id = "c28xIQaWQ6MzM5MzczMDA2NzMzMzkwNzIwNA"
+
 
 def crewai_integration():
-    print("CrewAI integration")
+    toolset = StackOneToolSet()
+    tools = toolset.get_tools(
+        vertical="hris",
+        account_id=account_id,
+    )
+
+    # CrewAI uses LangChain tools natively
+    langchain_tools = tools.to_langchain()
+
+    agent = Agent(
+        role="HR Manager",
+        goal=f"What is the employee with the id {employee_id}?",
+        backstory="With over 10 years of experience in HR and employee management, "
+        "you excel at finding patterns in complex datasets.",
+        llm="gpt-4o-mini",
+        tools=langchain_tools,
+        max_iter=2,
+    )
+
+    task = Task(
+        description="What is the employee with the id c28xIQaWQ6MzM5MzczMDA2NzMzMzkwNzIwNA?",
+        agent=agent,
+        expected_output="A JSON object containing the employee's information",
+    )
+
+    crew = Crew(agents=[agent], tasks=[task])
+    print(crew.kickoff())
 
 
 if __name__ == "__main__":

examples/index.py

@@ -62,5 +62,11 @@ def quickstart():
 ## Next Steps
 
 Check out some examples:
+- [Error Handling](error-handling.md)
+- [StackOne Account IDs](stackone_account_ids.md)
+- [Available Tools](available_tools.md)
 - [OpenAI Integration](openai-integration.md)
+- [LangChain Integration](langchain-integration.md)
+- [CrewAI Integration](crewai-integration.md)
+- [LangGraph Tool Node](langgraph-tool-node.md)
 """

examples/langgraph_tool_node.py

@@ -7,6 +7,7 @@
 """
 
 
+# TODO: Add examples
 def langgraph_tool_node() -> None:
     print("LangGraph tool node")
 

packages/stackone-ai/stackone_ai/__init__.py

@@ -1,5 +1,5 @@
 """StackOne AI SDK"""
 
-from .tools import StackOneToolSet
+from .toolset import StackOneToolSet
 
 __all__ = ["StackOneToolSet"]

packages/stackone-ai/stackone_ai/models.py

@@ -1,11 +1,8 @@
-import base64
-import json
-from collections.abc import Callable, Sequence
-from typing import Annotated, Any, ClassVar
+from collections.abc import Sequence
+from typing import Any
 
-import requests
-from langchain_core.tools import BaseTool as LangChainBaseTool
-from pydantic import BaseModel, Field, PrivateAttr
+from langchain_core.tools import BaseTool
+from pydantic import BaseModel, Field
 
 
 class ExecuteConfig(BaseModel):
@@ -30,117 +27,23 @@ class ToolDefinition(BaseModel):
     execute: ExecuteConfig
 
 
-class BaseTool(BaseModel):
-    """Base Tool model with Pydantic validation"""
+class Tool(BaseModel):
+    """Base Tool model"""
 
+    name: str
     description: str
     parameters: ToolParameters
 
-    # Private attributes in Pydantic v2
-    _execute_config: ExecuteConfig = PrivateAttr()
-    _api_key: str = PrivateAttr()
-    _account_id: str | None = PrivateAttr(default=None)
-
-    def __init__(self, **data: Any) -> None:
-        super().__init__(**data)
-        execute_config = data.get("_execute_config")
-        api_key = data.get("_api_key")
-
-        if not isinstance(execute_config, ExecuteConfig):
-            raise ValueError("_execute_config must be an ExecuteConfig instance")
-        if not isinstance(api_key, str):
-            raise ValueError("_api_key must be a string")
-
-        self._execute_config = execute_config
-        self._api_key = api_key
-        self._account_id = data.get("_account_id")
-
     def execute(self, arguments: str | dict | None = None) -> dict[str, Any]:
-        """
-        Execute the tool with the given parameters
-
-        Args:
-            arguments: Either a JSON string or dict of arguments
-        """
-        # Handle both string and dict arguments
-        if isinstance(arguments, str):
-            kwargs = json.loads(arguments)
-        else:
-            kwargs = arguments or {}
-
-        url = self._execute_config.url
-        body_params = {}
-        query_params = {}
-        header_params = {}
-
-        # Separate parameters based on their OpenAPI location
-        if kwargs:
-            for key, value in kwargs.items():
-                param_location = self._execute_config.parameter_locations.get(key)
-
-                if param_location == "header":
-                    header_params[key] = str(value)
-                elif param_location == "query":
-                    query_params[key] = value
-                elif param_location == "path":
-                    url = url.replace(f"{{{key}}}", str(value))
-                elif param_location == "body":
-                    body_params[key] = value
-                else:
-                    # Default behavior for backward compatibility
-                    if f"{{{key}}}" in url:
-                        url = url.replace(f"{{{key}}}", str(value))
-                    elif self._execute_config.method.upper() in ["GET", "DELETE"]:
-                        query_params[key] = value
-                    else:
-                        body_params[key] = value
-
-        # Create basic auth header with API key as username
-        auth_string = base64.b64encode(f"{self._api_key}:".encode()).decode()
-
-        headers = {
-            "Authorization": f"Basic {auth_string}",
-            "User-Agent": "stackone-python/1.0.0",
-        }
-
-        if self._account_id:
-            headers["x-account-id"] = self._account_id
-
-        # Add custom header parameters
-        headers.update(header_params)
-
-        # Add predefined headers last to ensure they take precedence
-        headers.update(self._execute_config.headers)
-
-        request_kwargs: dict[str, Any] = {
-            "method": self._execute_config.method,
-            "url": url,
-            "headers": headers,
-        }
-
-        # Handle request body if we have body parameters
-        if body_params:
-            body_type = self._execute_config.body_type or "json"
-            if body_type == "json":
-                request_kwargs["json"] = body_params
-            elif body_type == "form":
-                request_kwargs["data"] = body_params
-
-        if query_params:
-            request_kwargs["params"] = query_params
-
-        response = requests.request(**request_kwargs)
-        response.raise_for_status()
-
-        result: dict[str, Any] = response.json()
-        return result
+        """Execute the tool with the given parameters"""
+        raise NotImplementedError
 
     def to_openai_function(self) -> dict:
         """Convert this tool to OpenAI's function format"""
         return {
             "type": "function",
             "function": {
-                "name": self._execute_config.name,
+                "name": self.name,
                 "description": self.description,
                 "parameters": {
                     "type": self.parameters.type,
@@ -152,46 +55,24 @@ def to_openai_function(self) -> dict:
             },
         }
 
-    @property
-    def name(self) -> str:
-        """Get the tool's name"""
-        return self._execute_config.name
-
-    def set_account_id(self, account_id: str | None) -> None:
-        """Set the account ID for this tool.
-
-        Args:
-            account_id: The account ID to use, or None to clear it
-        """
-        self._account_id = account_id
-
-    def get_account_id(self) -> str | None:
-        """Get the current account ID for this tool."""
-        return self._account_id
-
 
 class Tools:
     """Container for Tool instances"""
 
-    def __init__(self, tools: list[BaseTool]):
+    def __init__(self, tools: list[Tool]):
         self.tools = tools
-        # Create a name -> tool mapping for faster lookups
         self._tool_map = {tool.name: tool for tool in tools}
 
-    def __getitem__(self, index: int) -> BaseTool:
+    def __getitem__(self, index: int) -> Tool:
         return self.tools[index]
 
     def __len__(self) -> int:
         return len(self.tools)
 
-    def get_tool(self, name: str) -> BaseTool | None:
+    def get_tool(self, name: str) -> Tool | None:
         """Get a tool by its name"""
         return self._tool_map.get(name)
 
-    def to_openai(self) -> list[dict]:
-        """Convert all tools to OpenAI function format"""
-        return [tool.to_openai_function() for tool in self.tools]
-
     def set_account_id(self, account_id: str | None) -> None:
         """Set the account ID for all tools in this collection.
 
@@ -201,81 +82,18 @@ def set_account_id(self, account_id: str | None) -> None:
         for tool in self.tools:
             tool.set_account_id(account_id)
 
-    def to_langchain(self) -> Sequence[LangChainBaseTool]:
-        """Convert all tools to LangChain tool format"""
-        langchain_tools = []
-
+    def get_account_id(self) -> str | None:
+        """Get the current account ID for this tool."""
         for tool in self.tools:
-            # Create properly annotated schema for the tool
-            schema_props: dict[str, Field] = {}
-            annotations: dict[str, Any] = {}
-
-            for name, details in tool.parameters.properties.items():
-                # Convert OpenAPI types to Python types
-                python_type: type = str  # Default to str
-                if isinstance(details, dict):  # Check if details is a dict
-                    type_str = details.get("type", "string")
-                    if type_str == "number":
-                        python_type = float
-                    elif type_str == "integer":
-                        python_type = int
-                    elif type_str == "boolean":
-                        python_type = bool
+            account_id = tool.get_account_id()
+            if isinstance(account_id, str):  # Type guard to ensure we return str | None
+                return account_id
+        return None
 
-                    # Create Field with description if available
-                    field = Field(description=details.get("description", ""))
-                else:
-                    # Handle case where details is a string
-                    field = Field(description="")
-
-                schema_props[name] = field
-                annotations[name] = Annotated[python_type, field]
-
-            # Create the schema class with proper annotations
-            schema_class = type(
-                f"{tool.name.title()}Args",
-                (BaseModel,),
-                {
-                    "__annotations__": annotations,
-                    "__module__": __name__,
-                    **schema_props,
-                },
-            )
-
-            # Create the LangChain tool with proper type annotations
-            tool_annotations = {
-                "name": ClassVar[str],
-                "description": ClassVar[str],
-                "args_schema": ClassVar[type],
-            }
-
-            def create_run_method(t: BaseTool) -> Callable[..., Any]:
-                def _run(self: Any, **kwargs: Any) -> Any:
-                    # Convert kwargs to dict for execution
-                    return t.execute(kwargs)
-
-                return _run
-
-            def create_arun_method(t: BaseTool) -> Callable[..., Any]:
-                async def _arun(self: Any, **kwargs: Any) -> Any:
-                    # Convert kwargs to dict for execution
-                    return t.execute(kwargs)
-
-                return _arun
-
-            langchain_tool = type(
-                f"StackOne{tool.name.title()}Tool",
-                (LangChainBaseTool,),
-                {
-                    "__annotations__": tool_annotations,
-                    "__module__": __name__,
-                    "name": tool.name,
-                    "description": tool.description,
-                    "args_schema": schema_class,
-                    "_run": create_run_method(tool),
-                    "_arun": create_arun_method(tool),
-                },
-            )
-            langchain_tools.append(langchain_tool())
+    def to_openai(self) -> list[dict]:
+        """Convert all tools to OpenAI function format"""
+        return [tool.to_openai_function() for tool in self.tools]
 
-        return langchain_tools
+    def to_langchain(self) -> Sequence[BaseTool]:
+        """Convert all tools to LangChain format"""
+        return [tool.to_langchain() for tool in self.tools]