buddie-chat/buddie_chat.py

#!/usr/bin/env python3
"""
Buddie Chat - Aplicação nativa Linux para interação com MCPs
Descomplicar® Crescimento Digital
https://descomplicar.pt
"""

import gi
gi.require_version('Gtk', '4.0')
gi.require_version('Adw', '1')

from gi.repository import Gtk, Adw, GLib, Gio, GObject
import json
import os
import asyncio
import threading
from pathlib import Path
from typing import Dict, List, Optional, Any
import subprocess
import openai
from datetime import datetime

class MCPManager:
    """Gestor de conexões MCP"""
    
    def __init__(self):
        self.connections: Dict[str, Dict] = {}
        self.config_file = Path.home() / '.buddie-chat' / 'config.json'
        self.config_file.parent.mkdir(exist_ok=True)
        
    def load_config(self) -> Dict:
        """Carrega configuração"""
        if self.config_file.exists():
            with open(self.config_file, 'r') as f:
                return json.load(f)
        return {
            'openai_key': '',
            'openrouter_key': '',
            'mcps': [],
            'theme': 'dark'
        }
    
    def save_config(self, config: Dict):
        """Guarda configuração"""
        with open(self.config_file, 'w') as f:
            json.dump(config, f, indent=2)
    
    async def test_mcp_connection(self, mcp_config: Dict) -> bool:
        """Testa conexão MCP"""
        try:
            process = await asyncio.create_subprocess_exec(
                mcp_config['command'],
                *mcp_config.get('args', []),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            # Timeout de 5 segundos para teste
            try:
                stdout, stderr = await asyncio.wait_for(
                    process.communicate(b'{"jsonrpc": "2.0", "method": "initialize", "id": 1}\n'),
                    timeout=5.0
                )
                return process.returncode == 0
            except asyncio.TimeoutError:
                process.kill()
                return False
                
        except Exception as e:
            print(f"Erro ao testar MCP {mcp_config['name']}: {e}")
            return False
    
    async def send_chat_message(self, message: str, mcp_id: str, openai_key: str, openrouter_key: str = None) -> Dict:
        """Processa mensagem de chat"""
        try:
            client = openai.OpenAI(
                api_key=openai_key if not openrouter_key else openrouter_key,
                base_url="https://openrouter.ai/api/v1" if openrouter_key else None
            )
            
            model = "anthropic/claude-3.5-sonnet" if openrouter_key else "gpt-4o-mini"
            
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": message}],
                max_tokens=2000
            )
            
            return {
                'content': response.choices[0].message.content,
                'tool_calls': [],
                'success': True
            }
            
        except Exception as e:
            return {
                'content': f'Erro: {str(e)}',
                'tool_calls': [],
                'success': False
            }

class BuddieChat(Adw.Application):
    """Aplicação principal"""
    
    def __init__(self):
        super().__init__(application_id='pt.descomplicar.buddiechat')
        self.mcp_manager = MCPManager()
        self.config = self.mcp_manager.load_config()
        self.connect('activate', self.on_activate)
        
    def on_activate(self, app):
        """Ativa a aplicação"""
        self.window = BuddieChatWindow(self)
        self.window.set_application(self)
        self.window.present()

class BuddieChatWindow(Adw.ApplicationWindow):
    """Janela principal"""
    
    def __init__(self, app):
        super().__init__()
        self.app = app
        self.messages = []
        self.active_mcp = None
        
        self.set_title("Buddie Chat")
        self.set_default_size(1200, 800)
        
        # Layout principal
        self.main_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL)
        self.set_content(self.main_box)
        
        self.create_sidebar()
        self.create_main_content()
        
    def create_sidebar(self):
        """Cria barra lateral"""
        sidebar = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
        sidebar.set_size_request(350, -1)
        sidebar.add_css_class('sidebar')
        
        # Header
        header = Adw.HeaderBar()
        header.set_title_widget(Gtk.Label(label="Buddie Chat"))
        sidebar.append(header)
        
        # Configurações OpenAI/OpenRouter
        config_group = Adw.PreferencesGroup()
        config_group.set_title("Configuração de API")
        
        self.openai_entry = Adw.EntryRow()
        self.openai_entry.set_title("Chave OpenAI")
        self.openai_entry.set_text(self.app.config.get('openai_key', ''))
        self.openai_entry.connect('changed', self.on_config_changed)
        config_group.add(self.openai_entry)
        
        self.openrouter_entry = Adw.EntryRow()
        self.openrouter_entry.set_title("Chave OpenRouter")
        self.openrouter_entry.set_text(self.app.config.get('openrouter_key', ''))
        self.openrouter_entry.connect('changed', self.on_config_changed)
        config_group.add(self.openrouter_entry)
        
        sidebar.append(config_group)
        
        # Lista MCPs
        mcp_group = Adw.PreferencesGroup()
        mcp_group.set_title("Servidores MCP")
        
        # Botão adicionar MCP
        add_mcp_button = Gtk.Button(label="Adicionar MCP")
        add_mcp_button.add_css_class('suggested-action')
        add_mcp_button.connect('clicked', self.on_add_mcp)
        mcp_group.set_header_suffix(add_mcp_button)
        
        self.mcp_list = Gtk.ListBox()
        self.mcp_list.set_selection_mode(Gtk.SelectionMode.SINGLE)
        self.mcp_list.connect('row-selected', self.on_mcp_selected)
        mcp_group.add(self.mcp_list)
        
        sidebar.append(mcp_group)
        
        # Scrolled window para sidebar
        sidebar_scroll = Gtk.ScrolledWindow()
        sidebar_scroll.set_child(sidebar)
        sidebar_scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
        
        self.main_box.append(sidebar_scroll)
        
        self.update_mcp_list()
        
    def create_main_content(self):
        """Cria área principal de chat"""
        main_content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
        main_content.set_hexpand(True)
        
        # Header do chat
        self.chat_header = Adw.HeaderBar()
        self.chat_title = Gtk.Label(label="Selecione um MCP")
        self.chat_header.set_title_widget(self.chat_title)
        main_content.append(self.chat_header)
        
        # Área de mensagens
        self.messages_scroll = Gtk.ScrolledWindow()
        self.messages_scroll.set_vexpand(True)
        self.messages_scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
        
        self.messages_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
        self.messages_box.set_margin_top(12)
        self.messages_box.set_margin_bottom(12)
        self.messages_box.set_margin_start(12)
        self.messages_box.set_margin_end(12)
        
        self.messages_scroll.set_child(self.messages_box)
        main_content.append(self.messages_scroll)
        
        # Área de input
        input_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12)
        input_box.set_margin_top(12)
        input_box.set_margin_bottom(12)
        input_box.set_margin_start(12)
        input_box.set_margin_end(12)
        
        self.message_entry = Gtk.Entry()
        self.message_entry.set_hexpand(True)
        self.message_entry.set_placeholder_text("Digite sua mensagem...")
        self.message_entry.connect('activate', self.on_send_message)
        input_box.append(self.message_entry)
        
        self.send_button = Gtk.Button(label="Enviar")
        self.send_button.add_css_class('suggested-action')
        self.send_button.connect('clicked', self.on_send_message)
        input_box.append(self.send_button)
        
        main_content.append(input_box)
        
        self.main_box.append(main_content)
        
        self.show_welcome_message()
        
    def show_welcome_message(self):
        """Mostra mensagem de boas-vindas"""
        welcome_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
        welcome_box.set_valign(Gtk.Align.CENTER)
        welcome_box.set_halign(Gtk.Align.CENTER)
        
        icon = Gtk.Image.new_from_icon_name("system-users")
        icon.set_pixel_size(64)
        welcome_box.append(icon)
        
        title = Gtk.Label(label="Bem-vindo ao Buddie Chat!")
        title.add_css_class('title-1')
        welcome_box.append(title)
        
        subtitle = Gtk.Label(label="Configure suas chaves API e servidores MCP para começar")
        subtitle.add_css_class('dim-label')
        welcome_box.append(subtitle)
        
        self.messages_box.append(welcome_box)
        
    def update_mcp_list(self):
        """Atualiza lista de MCPs"""
        # Limpa lista atual
        while True:
            row = self.mcp_list.get_row_at_index(0)
            if row is None:
                break
            self.mcp_list.remove(row)
        
        # Adiciona MCPs da configuração
        for mcp in self.app.config.get('mcps', []):
            row = Adw.ActionRow()
            row.set_title(mcp['name'])
            row.set_subtitle(mcp['command'])
            row.mcp_data = mcp
            
            # Status indicator
            status_icon = Gtk.Image.new_from_icon_name("network-wireless-signal-good")
            status_icon.add_css_class('success')
            row.add_suffix(status_icon)
            
            self.mcp_list.append(row)
        
    def on_mcp_selected(self, listbox, row):
        """Seleciona MCP"""
        if row:
            self.active_mcp = row.mcp_data
            self.chat_title.set_label(f"Chat - {self.active_mcp['name']}")
            self.clear_messages()
            
    def on_config_changed(self, entry):
        """Salva alterações na configuração"""
        self.app.config['openai_key'] = self.openai_entry.get_text()
        self.app.config['openrouter_key'] = self.openrouter_entry.get_text()
        self.app.mcp_manager.save_config(self.app.config)
        
    def on_add_mcp(self, button):
        """Adiciona novo MCP"""
        dialog = MCPConfigDialog(self)
        dialog.present()
        
    def add_mcp_to_config(self, mcp_data):
        """Adiciona MCP à configuração"""
        if 'mcps' not in self.app.config:
            self.app.config['mcps'] = []
        
        self.app.config['mcps'].append(mcp_data)
        self.app.mcp_manager.save_config(self.app.config)
        self.update_mcp_list()
        
    def on_send_message(self, widget):
        """Envia mensagem"""
        message = self.message_entry.get_text().strip()
        if not message or not self.active_mcp:
            return
            
        self.message_entry.set_text("")
        self.add_user_message(message)
        
        # Processa mensagem em thread separada
        threading.Thread(
            target=self.process_message_async,
            args=(message,),
            daemon=True
        ).start()
        
    def process_message_async(self, message):
        """Processa mensagem de forma assíncrona"""
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        try:
            response = loop.run_until_complete(
                self.app.mcp_manager.send_chat_message(
                    message,
                    self.active_mcp['id'],
                    self.app.config.get('openai_key', ''),
                    self.app.config.get('openrouter_key', '')
                )
            )
            
            GLib.idle_add(self.add_assistant_message, response['content'])
            
        except Exception as e:
            GLib.idle_add(self.add_assistant_message, f"Erro: {str(e)}")
        finally:
            loop.close()
            
    def add_user_message(self, message):
        """Adiciona mensagem do utilizador"""
        self.clear_welcome()
        
        message_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL)
        message_box.set_halign(Gtk.Align.END)
        message_box.set_margin_bottom(6)
        
        bubble = Gtk.Label(label=message)
        bubble.set_wrap(True)
        bubble.set_margin_top(12)
        bubble.set_margin_bottom(12)
        bubble.set_margin_start(16)
        bubble.set_margin_end(16)
        bubble.add_css_class('user-message')
        
        message_box.append(bubble)
        self.messages_box.append(message_box)
        
        # Scroll para baixo
        adj = self.messages_scroll.get_vadjustment()
        GLib.idle_add(lambda: adj.set_value(adj.get_upper()))
        
    def add_assistant_message(self, message):
        """Adiciona mensagem do assistente"""
        message_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL)
        message_box.set_halign(Gtk.Align.START)
        message_box.set_margin_bottom(6)
        
        bubble = Gtk.Label(label=message)
        bubble.set_wrap(True)
        bubble.set_margin_top(12)
        bubble.set_margin_bottom(12)
        bubble.set_margin_start(16)
        bubble.set_margin_end(16)
        bubble.add_css_class('assistant-message')
        
        message_box.append(bubble)
        self.messages_box.append(message_box)
        
        # Scroll para baixo
        adj = self.messages_scroll.get_vadjustment()
        GLib.idle_add(lambda: adj.set_value(adj.get_upper()))
        
    def clear_welcome(self):
        """Remove mensagem de boas-vindas"""
        if self.messages_box.get_first_child():
            child = self.messages_box.get_first_child()
            if isinstance(child, Gtk.Box) and child.get_valign() == Gtk.Align.CENTER:
                self.messages_box.remove(child)
                
    def clear_messages(self):
        """Limpa todas as mensagens"""
        while True:
            child = self.messages_box.get_first_child()
            if child is None:
                break
            self.messages_box.remove(child)

class MCPConfigDialog(Adw.Window):
    """Diálogo de configuração MCP"""
    
    def __init__(self, parent):
        super().__init__()
        self.parent_window = parent
        self.set_title("Configurar MCP")
        self.set_default_size(500, 400)
        self.set_transient_for(parent)
        self.set_modal(True)
        
        # Layout principal
        main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
        self.set_content(main_box)
        
        # Header
        header = Adw.HeaderBar()
        cancel_button = Gtk.Button(label="Cancelar")
        cancel_button.connect('clicked', lambda x: self.close())
        header.pack_start(cancel_button)
        
        save_button = Gtk.Button(label="Guardar")
        save_button.add_css_class('suggested-action')
        save_button.connect('clicked', self.on_save)
        header.pack_end(save_button)
        
        main_box.append(header)
        
        # Formulário
        form = Adw.PreferencesPage()
        
        group = Adw.PreferencesGroup()
        group.set_title("Configuração do Servidor MCP")
        
        self.name_entry = Adw.EntryRow()
        self.name_entry.set_title("Nome")
        group.add(self.name_entry)
        
        self.command_entry = Adw.EntryRow()
        self.command_entry.set_title("Comando")
        group.add(self.command_entry)
        
        self.args_entry = Adw.EntryRow()
        self.args_entry.set_title("Argumentos (separados por espaço)")
        group.add(self.args_entry)
        
        form.add(group)
        
        # Scroll
        scroll = Gtk.ScrolledWindow()
        scroll.set_child(form)
        scroll.set_vexpand(True)
        main_box.append(scroll)
        
    def on_save(self, button):
        """Guarda configuração MCP"""
        name = self.name_entry.get_text().strip()
        command = self.command_entry.get_text().strip()
        args = self.args_entry.get_text().strip().split() if self.args_entry.get_text().strip() else []
        
        if not name or not command:
            return
            
        mcp_data = {
            'id': f"mcp_{len(self.parent_window.app.config.get('mcps', []))}",
            'name': name,
            'command': command,
            'args': args,
            'enabled': True
        }
        
        self.parent_window.add_mcp_to_config(mcp_data)
        self.close()

def main():
    """Função principal"""
    app = BuddieChat()
    return app.run(None)

if __name__ == '__main__':
    main()