leocrm/app/plugins/builtins/automation/agent_comm.py

"""Agent-to-Agent communication router."""

from __future__ import annotations

import json
import logging
import uuid
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)


async def send_agent_message(
    from_agent_id: str,
    to_agent_name: str,
    message: str,
    db: AsyncSession,
    tenant_id: uuid.UUID,
) -> dict:
    """Send a message from one agent to another.

    1. Find target agent by name
    2. Create a kommunikation message in a dedicated agent room
    3. Enqueue run_agent for the target agent with the message as trigger_data
    4. Return delivery status
    """
    from app.plugins.builtins.automation.models import AgentDefinition
    from app.plugins.builtins.automation.agent_runner import run_agent

    # 1. Find target agent by name
    result = await db.execute(
        select(AgentDefinition)
        .where(AgentDefinition.name == to_agent_name)
        .where(AgentDefinition.tenant_id == tenant_id)
        .limit(1)
    )
    target_agent = result.scalar_one_or_none()

    if target_agent is None:
        logger.warning(
            "Target agent '%s' not found for message from agent %s",
            to_agent_name, from_agent_id,
        )
        return {"status": "error", "error": f"Target agent '{to_agent_name}' not found"}

    if not target_agent.is_active:
        logger.warning(
            "Target agent '%s' is inactive, cannot deliver message from %s",
            to_agent_name, from_agent_id,
        )
        return {"status": "error", "error": f"Target agent '{to_agent_name}' is inactive"}

    # 2. Create a kommunikation message in a dedicated agent room
    try:
        from app.plugins.builtins.kommunikation.models import Message, Room
        from app.plugins.builtins.kommunikation.services import RoomService

        # Find or create the agent-to-agent room
        room_name = f"agent:{from_agent_id}:{target_agent.id}"
        room_result = await db.execute(
            select(Room).where(Room.name == room_name).limit(1)
        )
        room = room_result.scalar_one_or_none()

        if room is None:
            # Create a new room for agent communication
            room = Room(
                tenant_id=tenant_id,
                name=room_name,
                display_name=f"Agent Chat: {from_agent_id} -> {to_agent_name}",
                room_type="agent_comm",
                is_direct=True,
            )
            db.add(room)
            await db.flush()

        # Create the message
        msg = Message(
            tenant_id=tenant_id,
            room_id=room.id,
            sender_id=from_agent_id,
            sender_type="agent",
            content=message,
            message_type="agent_comm",
        )
        db.add(msg)
        await db.flush()

        logger.info(
            "Agent message created: %s -> %s in room %s",
            from_agent_id, to_agent_name, room_name,
        )
    except ImportError:
        logger.warning("Kommunikation plugin not available, skipping message storage")
    except Exception as e:
        logger.exception("Failed to create kommunikation message: %s", e)

    # 3. Enqueue run_agent for the target agent with the message as trigger_data
    try:
        trigger_data = {
            "from_agent_id": from_agent_id,
            "message": message,
            "type": "agent_comm",
        }
        # Run the target agent with the message as trigger data
        result_data = await run_agent(
            ctx={},
            agent_id=str(target_agent.id),
            trigger_type="agent_comm",
            trigger_data=trigger_data,
        )
        logger.info(
            "Target agent %s executed with message from %s: status=%s",
            to_agent_name, from_agent_id, result_data.get("status"),
        )
    except Exception as e:
        logger.exception("Failed to run target agent %s: %s", to_agent_name, e)
        return {"status": "error", "error": f"Failed to activate target agent: {e}"}

    return {
        "status": "sent",
        "target_agent": to_agent_name,
        "target_agent_id": str(target_agent.id),
    }


def register_agent_comm_tool():
    """Register the send_agent_message tool in the global tool registry."""
    from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry

    registry = get_tool_registry()

    async def handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
        """Handle send_agent_message tool call from an AI agent."""
        from app.core.db import get_session_factory

        to_agent_name = arguments.get("to_agent_name", "")
        message = arguments.get("message", "")
        from_agent_id = context.get("agent_id", "unknown")
        tenant_id_str = context.get("tenant_id", "")

        if not to_agent_name or not message:
            return json.dumps({"status": "error", "error": "Missing to_agent_name or message"})

        try:
            tenant_id = uuid.UUID(tenant_id_str) if tenant_id_str else uuid.uuid4()
        except (ValueError, TypeError):
            return json.dumps({"status": "error", "error": "Invalid tenant_id"})

        factory = get_session_factory()
        async with factory() as db:
            result = await send_agent_message(
                from_agent_id=from_agent_id,
                to_agent_name=to_agent_name,
                message=message,
                db=db,
                tenant_id=tenant_id,
            )
        return json.dumps(result)

    registry.register(
        name="send_agent_message",
        description="Send a message to another agent. The target agent will be activated with your message.",
        parameters={
            "type": "object",
            "properties": {
                "to_agent_name": {
                    "type": "string",
                    "description": "Name of the target agent",
                },
                "message": {
                    "type": "string",
                    "description": "Message to send",
                },
            },
            "required": ["to_agent_name", "message"],
        },
        handler=handler,
        plugin_name="automation",
        required_permission="agents:execute",
        category="communication",
    )
    logger.info("Agent communication tool 'send_agent_message' registered")


def unregister_agent_comm_tool():
    """Unregister the send_agent_message tool."""
    from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry

    registry = get_tool_registry()
    registry.unregister("send_agent_message")
    logger.info("Agent communication tool 'send_agent_message' unregistered")