Communication API

from marsys.coordination.communication import CommunicationManager
from marsys.coordination.config import CommunicationConfig

CommunicationManager(
    config: Optional[CommunicationConfig] = None
)

def register_channel(channel: CommunicationChannel) -> None

async def request_user_input(
    prompt: str,
    session_id: str,
    channel_id: Optional[str] = None,
    timeout: Optional[float] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> str

async def send_message(
    message: str,
    session_id: str,
    channel_id: Optional[str] = None,
    message_type: str = "info"
) -> None

# Initialize manager
config = CommunicationConfig(
    use_rich_formatting=True,
    theme_name="modern"
)
manager = CommunicationManager(config)

# Request user input
response = await manager.request_user_input(
    prompt="What would you like to analyze?",
    session_id="session_123"
)

# Send message
await manager.send_message(
    message="Analysis complete!",
    session_id="session_123",
    message_type="success"
)

from marsys.coordination.communication import UserNodeHandler

async def handle_user_node(
    branch: ExecutionBranch,
    incoming_message: Any,
    context: Dict[str, Any]
) -> StepResult

handler = UserNodeHandler(communication_manager)

# Handle user interaction
result = await handler.handle_user_node(
    branch=current_branch,
    incoming_message="Please approve the analysis results",
    context={"session_id": "session_123"}
)

# Process result
if result.success:
    user_response = result.data["response"]
    next_agent = result.data.get("next_agent")

from marsys.coordination.communication.channels import TerminalChannel

channel = TerminalChannel()

# Send message
await channel.send("Enter your choice:")

# Receive input
response = await channel.receive(timeout=30.0)

from marsys.coordination.communication.channels import EnhancedTerminalChannel

channel = EnhancedTerminalChannel(
    theme_name="modern",
    use_rich=True
)

# Send formatted message
await channel.send({
    "content": "## Analysis Results",
    "format": "markdown",
    "style": "success"
})

from marsys.coordination.communication.channels import PrefixedCLIChannel

channel = PrefixedCLIChannel(
    prefix_width=25,
    show_timestamps=True
)

# Send with prefix
await channel.send(
    message="Processing data...",
    metadata={"agent_name": "DataProcessor"}
)
# Output: [DataProcessor]     Processing data...

# In topology
topology = {
    "agents": ["Agent1", "User", "Agent2"],
    "flows": [
        "Agent1 -> User",
        "User -> Agent2"
    ]
}

# Agent1 response triggers user interaction
response = {
    "next_action": "invoke_agent",
    "action_input": "User",
    "message": "Please review the results"
}

# User sees message and provides input
# System routes response to Agent2

# Agent triggers error recovery
response = {
    "next_action": "error_recovery",
    "error_details": {
        "type": "api_quota_exceeded",
        "message": "API quota exceeded"
    },
    "suggested_action": "retry_with_different_model"
}

# Routes to User node for intervention
# User can:
# - Retry with same settings
# - Change model
# - Skip operation
# - Abort workflow

# Configure approval interaction
interaction = UserInteraction(
    id="approval_123",
    prompt="Approve deployment to production?",
    options=["approve", "reject", "review"],
    metadata={
        "changes": ["Update API", "Database migration"],
        "risk_level": "medium"
    }
)

# Request approval
response = await manager.request_structured_input(interaction)

if response == "approve":
    # Continue with deployment
    pass
elif response == "review":
    # Show detailed changes
    pass

from marsys.coordination.communication import EventBus

bus = EventBus()

# Subscribe to events
def on_input(data):
    print(f"User input: {data['response']}")

bus.subscribe("user.input.received", on_input)

# Publish event
await bus.publish(
    "user.input.received",
    {"response": "yes", "session_id": "123"}
)

from marsys.coordination.communication import CommunicationChannel

class WebSocketChannel(CommunicationChannel):
    """WebSocket-based communication channel."""

    def __init__(self, ws_url: str):
        super().__init__(channel_id="websocket")
        self.ws_url = ws_url
        self.ws = None

    async def connect(self):
        self.ws = await websocket.connect(self.ws_url)

    async def send(self, message: str, metadata: Dict = None):
        await self.ws.send(json.dumps({
            "message": message,
            "metadata": metadata
        }))

    async def receive(self, timeout: float = None):
        try:
            data = await asyncio.wait_for(
                self.ws.receive(),
                timeout=timeout
            )
            return json.loads(data)["response"]
        except asyncio.TimeoutError:
            return None

    def is_available(self) -> bool:
        return self.ws and not self.ws.closed

    async def close(self):
        if self.ws:
            await self.ws.close()