gateway/modules/serviceCenter/services/serviceAgent/toolboxRegistry.py

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Toolbox Registry for the Agent service.
Manages thematic tool groupings (toolboxes) and the `requestToolbox` meta-tool.
"""

import logging
from typing import Dict, List, Any, Optional, Set
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)


class ToolboxDefinition(BaseModel):
    """Definition of a thematic toolbox."""
    id: str = Field(description="Unique toolbox identifier (e.g. 'core', 'email', 'workflow')")
    label: str = Field(description="Human-readable label")
    description: str = Field(default="", description="What this toolbox provides")
    featureCode: Optional[str] = Field(default=None, description="Feature code if toolbox is feature-specific")
    tools: List[str] = Field(default_factory=list, description="Tool names belonging to this toolbox")
    isDefault: bool = Field(default=False, description="If true, toolbox is active by default")
    requiresConnection: Optional[str] = Field(
        default=None,
        description="Connection authority required (e.g. 'microsoft', 'google'). None = always available."
    )


class ToolboxRegistry:
    """Registry for toolbox definitions. Manages activation and tool lookup."""

    def __init__(self):
        self._toolboxes: Dict[str, ToolboxDefinition] = {}

    def registerToolbox(self, toolbox: ToolboxDefinition) -> None:
        """Register a toolbox definition."""
        if toolbox.id in self._toolboxes:
            logger.debug("Toolbox '%s' already registered, updating", toolbox.id)
        self._toolboxes[toolbox.id] = toolbox
        logger.debug("Registered toolbox: %s (%d tools, default=%s)", toolbox.id, len(toolbox.tools), toolbox.isDefault)

    def getToolbox(self, toolboxId: str) -> Optional[ToolboxDefinition]:
        """Get a toolbox by ID."""
        return self._toolboxes.get(toolboxId)

    def getAllToolboxes(self) -> List[ToolboxDefinition]:
        """Get all registered toolboxes."""
        return list(self._toolboxes.values())

    def getDefaultToolboxes(self) -> List[ToolboxDefinition]:
        """Get all default toolboxes (active at agent start)."""
        return [tb for tb in self._toolboxes.values() if tb.isDefault]

    def getActiveToolboxes(self, userConnections: List[str] = None) -> List[ToolboxDefinition]:
        """
        Get toolboxes available to the user based on their connections.
        Toolboxes without requiresConnection are always available.
        Toolboxes with requiresConnection are available only if the user has that connection.
        """
        available = []
        connectionAuthorities: Set[str] = set(userConnections or [])
        for tb in self._toolboxes.values():
            if tb.requiresConnection is None:
                available.append(tb)
            elif tb.requiresConnection in connectionAuthorities:
                available.append(tb)
        return available

    def getToolsForToolboxes(self, toolboxIds: List[str]) -> List[str]:
        """Get the union of all tool names for the given toolbox IDs."""
        tools: Set[str] = set()
        for tbId in toolboxIds:
            tb = self._toolboxes.get(tbId)
            if tb:
                tools.update(tb.tools)
        return sorted(tools)

    def getToolboxForTool(self, toolName: str) -> Optional[str]:
        """Find which toolbox a tool belongs to."""
        for tb in self._toolboxes.values():
            if toolName in tb.tools:
                return tb.id
        return None

    def toApiResponse(self, userConnections: List[str] = None) -> List[Dict[str, Any]]:
        """Serialize available toolboxes for API response."""
        available = self.getActiveToolboxes(userConnections)
        return [
            {
                "id": tb.id,
                "label": tb.label,
                "description": tb.description,
                "toolCount": len(tb.tools),
                "isDefault": tb.isDefault,
                "requiresConnection": tb.requiresConnection,
            }
            for tb in available
        ]


# Module-level singleton
_toolboxRegistry = ToolboxRegistry()


def getToolboxRegistry() -> ToolboxRegistry:
    """Get the global toolbox registry singleton."""
    return _toolboxRegistry


def _registerDefaultToolboxes() -> None:
    """Register the default set of toolboxes."""
    defaults = [
        ToolboxDefinition(
            id="core",
            label="Core Tools",
            description="Basic agent tools: search, read, write, web",
            isDefault=True,
            tools=[],
        ),
        ToolboxDefinition(
            id="ai",
            label="AI Tools",
            description="AI-powered analysis and generation",
            isDefault=True,
            tools=[],
        ),
        ToolboxDefinition(
            id="datasources",
            label="Data Sources",
            description="Access external data sources and databases",
            isDefault=False,
            tools=[],
        ),
        ToolboxDefinition(
            id="email",
            label="Email",
            description="Read and send emails via Outlook/Gmail",
            requiresConnection="microsoft",
            isDefault=False,
            tools=[],
        ),
        ToolboxDefinition(
            id="sharepoint",
            label="SharePoint",
            description="Access SharePoint sites, lists, and files",
            requiresConnection="microsoft",
            isDefault=False,
            tools=[],
        ),
        ToolboxDefinition(
            id="clickup",
            label="ClickUp",
            description="Manage ClickUp tasks and projects",
            requiresConnection="clickup",
            isDefault=False,
            tools=[],
        ),
        ToolboxDefinition(
            id="jira",
            label="Jira",
            description="Manage Jira issues and projects",
            requiresConnection="jira",
            isDefault=False,
            tools=[],
        ),
        ToolboxDefinition(
            id="workflow",
            label="Workflow",
            description="Graph manipulation tools for the visual editor",
            featureCode="graphicalEditor",
            isDefault=False,
            tools=[
                "readWorkflowGraph", "addNode", "removeNode", "connectNodes",
                "setNodeParameter", "listAvailableNodeTypes", "validateGraph",
                "listWorkflowHistory", "readWorkflowMessages",
            ],
        ),
    ]
    for tb in defaults:
        _toolboxRegistry.registerToolbox(tb)


_registerDefaultToolboxes()