"""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")