#!/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()