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

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""ActionToolAdapter: wraps existing workflow actions (dynamicMode=True) as agent tools."""

import logging
from typing import Dict, Any, List, Optional

from modules.serviceCenter.services.serviceAgent.datamodelAgent import (
    ToolDefinition, ToolResult
)
from modules.serviceCenter.services.serviceAgent.toolRegistry import ToolRegistry

logger = logging.getLogger(__name__)


class ActionToolAdapter:
    """Wraps existing Workflow-Actions as Agent-Tools.

    Iterates over discovered methods, finds actions with dynamicMode=True,
    and registers them in the ToolRegistry with a compound name (method.action).
    """

    def __init__(self, actionExecutor):
        self._actionExecutor = actionExecutor
        self._registeredTools: List[str] = []

    def registerAll(self, toolRegistry: ToolRegistry):
        """Discover and register all dynamicMode actions as agent tools."""
        from modules.workflows.processing.shared.methodDiscovery import methods

        registered = 0
        for methodName, methodInfo in methods.items():
            if not methodName[0].isupper():
                continue

            shortName = methodName.replace("Method", "").lower()
            methodInstance = methodInfo["instance"]

            for actionName, actionInfo in methodInfo["actions"].items():
                actionDef = methodInstance._actions.get(actionName)
                if not actionDef or not getattr(actionDef, "dynamicMode", False):
                    continue

                compoundName = f"{shortName}_{actionName}"
                toolDef = _buildToolDefinition(compoundName, actionDef, actionInfo)

                handler = _createDispatchHandler(self._actionExecutor, shortName, actionName)
                toolRegistry.registerFromDefinition(toolDef, handler)
                self._registeredTools.append(compoundName)
                registered += 1

        logger.info(f"ActionToolAdapter: registered {registered} tools from workflow actions")

    @property
    def registeredTools(self) -> List[str]:
        """Names of all tools registered by this adapter."""
        return list(self._registeredTools)


def _buildToolDefinition(compoundName: str, actionDef, actionInfo: Dict[str, Any]) -> ToolDefinition:
    """Build a ToolDefinition from a WorkflowActionDefinition."""
    parameters = _convertParameterSchema(actionInfo.get("parameters", {}))

    return ToolDefinition(
        name=compoundName,
        description=actionDef.description or actionInfo.get("description", ""),
        parameters=parameters,
        readOnly=False
    )


def _convertParameterSchema(actionParams: Dict[str, Any]) -> Dict[str, Any]:
    """Convert workflow action parameter schema to JSON Schema for tool definitions."""
    properties = {}
    required = []

    for paramName, paramInfo in actionParams.items():
        paramType = paramInfo.get("type", "str") if isinstance(paramInfo, dict) else "str"
        paramDesc = paramInfo.get("description", "") if isinstance(paramInfo, dict) else ""
        paramRequired = paramInfo.get("required", False) if isinstance(paramInfo, dict) else False

        jsonType = _pythonTypeToJsonType(paramType)
        prop: Dict[str, Any] = {
            "type": jsonType,
            "description": paramDesc,
        }
        if jsonType == "array":
            prop["items"] = _pythonTypeToArrayItems(paramType) or {"type": "string"}
        properties[paramName] = prop

        if paramRequired:
            required.append(paramName)

    return {
        "type": "object",
        "properties": properties,
        "required": required
    }


_TYPE_MAPPING = {
    "str": "string",
    "int": "integer",
    "float": "number",
    "bool": "boolean",
    "list": "array",
    "dict": "object",
    "List[str]": "array",
    "List[int]": "array",
    "List[dict]": "array",
    "List[float]": "array",
    "Dict[str, Any]": "object",
}

_ARRAY_ITEMS_MAPPING = {
    "list": {"type": "string"},
    "List[str]": {"type": "string"},
    "List[int]": {"type": "integer"},
    "List[float]": {"type": "number"},
    "List[dict]": {"type": "object"},
}


def _pythonTypeToJsonType(pythonType: str) -> str:
    """Map Python type strings to JSON Schema types."""
    return _TYPE_MAPPING.get(pythonType, "string")


def _pythonTypeToArrayItems(pythonType: str) -> Optional[Dict[str, Any]]:
    """Return the JSON Schema `items` descriptor for array types, or None."""
    return _ARRAY_ITEMS_MAPPING.get(pythonType)


def _createDispatchHandler(actionExecutor, methodName: str, actionName: str):
    """Create an async handler that dispatches to the ActionExecutor."""
    async def _handler(args: Dict[str, Any], context: Dict[str, Any]) -> ToolResult:
        try:
            result = await actionExecutor.executeAction(methodName, actionName, args)
            data = _formatActionResult(result)
            return ToolResult(
                toolCallId="",
                toolName=f"{methodName}_{actionName}",
                success=result.success,
                data=data,
                error=result.error
            )
        except Exception as e:
            logger.error(f"ActionToolAdapter dispatch failed for {methodName}_{actionName}: {e}")
            return ToolResult(
                toolCallId="",
                toolName=f"{methodName}_{actionName}",
                success=False,
                error=str(e)
            )
    return _handler


def _formatActionResult(result) -> str:
    """Format an ActionResult into a text representation for the agent."""
    parts = []

    if result.resultLabel:
        parts.append(f"Result: {result.resultLabel}")

    if result.error:
        parts.append(f"Error: {result.error}")

    if result.documents:
        parts.append(f"Documents ({len(result.documents)}):")
        for doc in result.documents:
            docName = getattr(doc, "documentName", "unnamed")
            docType = getattr(doc, "mimeType", "unknown")
            parts.append(f"  - {docName} ({docType})")
            docData = getattr(doc, "documentData", None)
            if docData and isinstance(docData, str) and len(docData) < 2000:
                parts.append(f"    Content: {docData[:2000]}")

    if not parts:
        parts.append("Action completed successfully." if result.success else "Action failed.")

    return "\n".join(parts)