feat: allow list tools by tool type [PRO-870] (#4036)

* feat: allow list tools by tool type

* chore: update list

* chore: respond to comments

* chore: refactor tools hella

* Add tests to managers

* chore: branch

---------

Co-authored-by: Shubham Naik <[email protected]>
Co-authored-by: Matt Zhou <[email protected]>

Author: 3 people

letta/server/rest_api/routers/v1/tools.py

@@ -27,7 +27,7 @@
 from letta.orm.errors import UniqueConstraintViolationError
 from letta.orm.mcp_oauth import OAuthSessionStatus
 from letta.prompts.gpt_system import get_system_text
-from letta.schemas.enums import MessageRole
+from letta.schemas.enums import MessageRole, ToolType
 from letta.schemas.letta_message import ToolReturnMessage
 from letta.schemas.letta_message_content import TextContent
 from letta.schemas.mcp import UpdateSSEMCPServer, UpdateStdioMCPServer, UpdateStreamableHTTPMCPServer
@@ -62,16 +62,94 @@ async def delete_tool(
 
 @router.get("/count", response_model=int, operation_id="count_tools")
 async def count_tools(
+    name: Optional[str] = None,
+    names: Optional[List[str]] = Query(None, description="Filter by specific tool names"),
+    tool_ids: Optional[List[str]] = Query(
+        None, description="Filter by specific tool IDs - accepts repeated params or comma-separated values"
+    ),
+    search: Optional[str] = Query(None, description="Search tool names (case-insensitive partial match)"),
+    tool_types: Optional[List[str]] = Query(None, description="Filter by tool type(s) - accepts repeated params or comma-separated values"),
+    exclude_tool_types: Optional[List[str]] = Query(
+        None, description="Tool type(s) to exclude - accepts repeated params or comma-separated values"
+    ),
+    return_only_letta_tools: Optional[bool] = Query(False, description="Count only tools with tool_type starting with 'letta_'"),
+    exclude_letta_tools: Optional[bool] = Query(False, description="Exclude built-in Letta tools from the count"),
     server: SyncServer = Depends(get_letta_server),
     actor_id: Optional[str] = Header(None, alias="user_id"),
-    include_base_tools: Optional[bool] = Query(False, description="Include built-in Letta tools in the count"),
 ):
     """
     Get a count of all tools available to agents belonging to the org of the user.
     """
     try:
+        # Helper function to parse tool types - supports both repeated params and comma-separated values
+        def parse_tool_types(tool_types_input: Optional[List[str]]) -> Optional[List[str]]:
+            if tool_types_input is None:
+                return None
+
+            # Flatten any comma-separated values and validate against ToolType enum
+            flattened_types = []
+            for item in tool_types_input:
+                # Split by comma in case user provided comma-separated values
+                types_in_item = [t.strip() for t in item.split(",") if t.strip()]
+                flattened_types.extend(types_in_item)
+
+            # Validate each type against the ToolType enum
+            valid_types = []
+            valid_values = [tt.value for tt in ToolType]
+
+            for tool_type in flattened_types:
+                if tool_type not in valid_values:
+                    raise HTTPException(
+                        status_code=400, detail=f"Invalid tool_type '{tool_type}'. Must be one of: {', '.join(valid_values)}"
+                    )
+                valid_types.append(tool_type)
+
+            return valid_types if valid_types else None
+
+        # Parse and validate tool types (same logic as list_tools)
+        tool_types_str = parse_tool_types(tool_types)
+        exclude_tool_types_str = parse_tool_types(exclude_tool_types)
+
         actor = await server.user_manager.get_actor_or_default_async(actor_id=actor_id)
-        return await server.tool_manager.size_async(actor=actor, include_base_tools=include_base_tools)
+
+        # Combine single name with names list for unified processing (same logic as list_tools)
+        combined_names = []
+        if name is not None:
+            combined_names.append(name)
+        if names is not None:
+            combined_names.extend(names)
+
+        # Use None if no names specified, otherwise use the combined list
+        final_names = combined_names if combined_names else None
+
+        # Helper function to parse tool IDs - supports both repeated params and comma-separated values
+        def parse_tool_ids(tool_ids_input: Optional[List[str]]) -> Optional[List[str]]:
+            if tool_ids_input is None:
+                return None
+
+            # Flatten any comma-separated values
+            flattened_ids = []
+            for item in tool_ids_input:
+                # Split by comma in case user provided comma-separated values
+                ids_in_item = [id.strip() for id in item.split(",") if id.strip()]
+                flattened_ids.extend(ids_in_item)
+
+            return flattened_ids if flattened_ids else None
+
+        # Parse tool IDs (same logic as list_tools)
+        final_tool_ids = parse_tool_ids(tool_ids)
+
+        # Get the count of tools using unified query
+        return await server.tool_manager.count_tools_async(
+            actor=actor,
+            tool_types=tool_types_str,
+            exclude_tool_types=exclude_tool_types_str,
+            names=final_names,
+            tool_ids=final_tool_ids,
+            search=search,
+            return_only_letta_tools=return_only_letta_tools,
+            exclude_letta_tools=exclude_letta_tools,
+        )
     except Exception as e:
         print(f"Error occurred: {e}")
         raise HTTPException(status_code=500, detail=str(e))
@@ -99,20 +177,93 @@ async def list_tools(
     after: Optional[str] = None,
     limit: Optional[int] = 50,
     name: Optional[str] = None,
+    names: Optional[List[str]] = Query(None, description="Filter by specific tool names"),
+    tool_ids: Optional[List[str]] = Query(
+        None, description="Filter by specific tool IDs - accepts repeated params or comma-separated values"
+    ),
+    search: Optional[str] = Query(None, description="Search tool names (case-insensitive partial match)"),
+    tool_types: Optional[List[str]] = Query(None, description="Filter by tool type(s) - accepts repeated params or comma-separated values"),
+    exclude_tool_types: Optional[List[str]] = Query(
+        None, description="Tool type(s) to exclude - accepts repeated params or comma-separated values"
+    ),
+    return_only_letta_tools: Optional[bool] = Query(False, description="Return only tools with tool_type starting with 'letta_'"),
     server: SyncServer = Depends(get_letta_server),
     actor_id: Optional[str] = Header(None, alias="user_id"),  # Extract user_id from header, default to None if not present
 ):
     """
     Get a list of all tools available to agents belonging to the org of the user
     """
     try:
+        # Helper function to parse tool types - supports both repeated params and comma-separated values
+        def parse_tool_types(tool_types_input: Optional[List[str]]) -> Optional[List[str]]:
+            if tool_types_input is None:
+                return None
+
+            # Flatten any comma-separated values and validate against ToolType enum
+            flattened_types = []
+            for item in tool_types_input:
+                # Split by comma in case user provided comma-separated values
+                types_in_item = [t.strip() for t in item.split(",") if t.strip()]
+                flattened_types.extend(types_in_item)
+
+            # Validate each type against the ToolType enum
+            valid_types = []
+            valid_values = [tt.value for tt in ToolType]
+
+            for tool_type in flattened_types:
+                if tool_type not in valid_values:
+                    raise HTTPException(
+                        status_code=400, detail=f"Invalid tool_type '{tool_type}'. Must be one of: {', '.join(valid_values)}"
+                    )
+                valid_types.append(tool_type)
+
+            return valid_types if valid_types else None
+
+        # Parse and validate tool types
+        tool_types_str = parse_tool_types(tool_types)
+        exclude_tool_types_str = parse_tool_types(exclude_tool_types)
+
         actor = await server.user_manager.get_actor_or_default_async(actor_id=actor_id)
+
+        # Combine single name with names list for unified processing
+        combined_names = []
         if name is not None:
-            tool = await server.tool_manager.get_tool_by_name_async(tool_name=name, actor=actor)
-            return [tool] if tool else []
+            combined_names.append(name)
+        if names is not None:
+            combined_names.extend(names)
 
-        # Get the list of tools
-        return await server.tool_manager.list_tools_async(actor=actor, after=after, limit=limit)
+        # Use None if no names specified, otherwise use the combined list
+        final_names = combined_names if combined_names else None
+
+        # Helper function to parse tool IDs - supports both repeated params and comma-separated values
+        def parse_tool_ids(tool_ids_input: Optional[List[str]]) -> Optional[List[str]]:
+            if tool_ids_input is None:
+                return None
+
+            # Flatten any comma-separated values
+            flattened_ids = []
+            for item in tool_ids_input:
+                # Split by comma in case user provided comma-separated values
+                ids_in_item = [id.strip() for id in item.split(",") if id.strip()]
+                flattened_ids.extend(ids_in_item)
+
+            return flattened_ids if flattened_ids else None
+
+        # Parse tool IDs
+        final_tool_ids = parse_tool_ids(tool_ids)
+
+        # Get the list of tools using unified query
+        return await server.tool_manager.list_tools_async(
+            actor=actor,
+            after=after,
+            limit=limit,
+            tool_types=tool_types_str,
+            exclude_tool_types=exclude_tool_types_str,
+            names=final_names,
+            tool_ids=final_tool_ids,
+            search=search,
+            return_only_letta_tools=return_only_letta_tools,
+        )
     except Exception as e:
         # Log or print the full exception here for debugging
         print(f"Error occurred: {e}")