Function observer_message_handler

evi/evi-python-control-plane/main.py:123–163

Handle messages received from the observer connection. This callback processes messages received from the control plane observer connection. It receives the same event types and shapes as the reference Chat socket.

(message: dict)

async def observer_message_handler(message: dict) -> None:
    """Handle messages received from the observer connection.

    This callback processes messages received from the control plane observer
    connection. It receives the same event types and shapes as the reference
    Chat socket.
    """
    msg_type = message.get("type", "unknown")

    if msg_type == "chat_metadata":
        print(
            f"[OBSERVER] Chat ID: {message.get('chat_id')}, Chat Group ID: {message.get('chat_group_id')}"
        )
        # Uncomment the line below for full message details
        # print(f"[OBSERVER] Full message: {json.dumps(message, indent=2)}")
    elif msg_type in ["user_message", "assistant_message"]:
        role = message.get("message", {}).get("role", "unknown").upper()
        content = message.get("message", {}).get("content", "")
        print(f"[OBSERVER] {role}: {content}")
    elif msg_type == "audio_output":
        # Audio output messages contain large base64-encoded data
        # Only print a summary to avoid cluttering the terminal
        data_length = len(message.get("data", ""))
        is_final = message.get("is_final_chunk", False)
        print(f"[OBSERVER] Audio output: {data_length} bytes, final_chunk={is_final}")
        # Uncomment the line below to see full audio message (very verbose!)
        # print(f"[OBSERVER] Full message: {json.dumps(message, indent=2)}")
    elif msg_type in ["user_interruption", "assistant_end"]:
        # These are expected message types, just acknowledge them silently
        # Uncomment the line below to see these messages
        # print(f"[OBSERVER] Received: {msg_type}")
        pass
    elif msg_type == "error":
        error_code = message.get("code", "unknown")
        error_msg = message.get("message", "unknown error")
        print(f"[OBSERVER] Error ({error_code}): {error_msg}")
        print(f"[OBSERVER] Full message: {json.dumps(message, indent=2)}")
    else:
        print(f"[OBSERVER] Unknown message type: <{msg_type}>")
        # Uncomment the line below for full message details
        # print(f"[OBSERVER] Full message: {json.dumps(message, indent=2)}")


async def control_plane_demo(