Building a Pieces Copilot TUI - Part 3: Advanced Features & Integration
Note: This tutorial is part of the Pieces CLI project. We welcome contributions! Feel free to open issues, submit PRs, or suggest improvements.
Introduction
Welcome to Part 3! In Part 2, we built the core UI components. Now, we'll add the advanced features to create a fully-functional Pieces Copilot TUI.
What we built in Part 2:
- ✅ Message widget (
chat_message.py)
- ✅ Chat panel (
chat_panel.py)
- ✅ Input widget (
chat_input.py)
- ✅ Basic view (
chat_view.py- simple version)
- ✅ Main application (
app.py)
Step 1: Create the Conversations List
Let's build chats_list.py - the sidebar showing all conversations:
# chats_list.py
"""Conversations list panel widget."""
from typing import Optional
from textual.widgets import Static, Button
from textual.containers import VerticalScroll, Vertical
from textual.message import Message
from pieces_os_client import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat
class ChatSelected(Message):
"""Message emitted when a chat is selected."""
def __init__(self, chat: BasicChat):
super().__init__()
self.chat = chat
class NewChatRequested(Message):
"""Message emitted when new chat is requested."""
pass
class ChatItem(Static):
"""Widget representing a single chat in the list."""
DEFAULT_CSS = """
ChatItem {
width: 100%;
height: auto;
padding: 1 2;
margin-bottom: 1;
background: $surface;
border-left: solid $primary;
}
ChatItem:hover {
background: $panel;
border-left: solid $accent;
}
ChatItem.active {
background: $primary 30%;
border-left: thick $accent;
text-style: bold;
}
ChatItem .chat-title {
color: $text;
text-style: bold;
}
ChatItem .chat-summary {
color: $text-muted;
text-style: italic;
}
"""
def __init__(self, chat: BasicChat, **kwargs):
super().__init__(**kwargs)
self.chat = chat
def compose(self):
"""Compose the chat item."""
title = self.chat.name or "Untitled"
summary = self.chat.summary or ""
if len(summary) > 50:
summary = summary[:47] + "..."
yield Static(f"💬 {title}", classes="chat-title")
if summary:
yield Static(summary, classes="chat-summary")
def on_click(self) -> None:
"""Handle click event."""
self.post_message(ChatSelected(self.chat))
class ChatsList(Vertical):
"""Panel to display list of conversations."""
DEFAULT_CSS = """
ChatsList {
width: 25%;
height: 100%;
border: solid $primary;
background: $background;
padding: 1;
}
ChatsList:focus-within {
border: solid $accent;
}
ChatsList Button {
width: 100%;
margin-bottom: 1;
}
ChatsList VerticalScroll {
height: 1fr;
}
ChatsList .empty-state {
text-align: center;
color: $text-muted;
text-style: italic;
margin: 2;
}
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.border_title = "Conversations"
self.active_chat: Optional[BasicChat] = None
self._chat_items = {}
def compose(self):
"""Compose the chats list panel."""
yield Button("➕ New Chat", id="new-chat-btn")
yield VerticalScroll(id="chats-container")
def load_chats(self, pieces_client: PiecesClient):
"""Load chats from the API."""
try:
chats = pieces_client.copilot.chats()
container = self.query_one("#chats-container", VerticalScroll)
# Clear existing items
container.remove_children()
self._chat_items.clear()
if not chats:
container.mount(Static("No chats yet...", classes="empty-state"))
else:
for chat in chats:
chat_item = ChatItem(chat)
self._chat_items[chat.id] = chat_item
container.mount(chat_item)
except (ConnectionError, AttributeError) as e:
container = self.query_one("#chats-container", VerticalScroll)
container.mount(Static(f"❌ Failed to load chats: {str(e)}", classes="empty-state"))
def set_active_chat(self, chat: Optional[BasicChat]):
"""Set the active chat and update UI."""
# Remove active class from all items
for item in self._chat_items.values():
item.remove_class("active")
self.active_chat = chat
# Add active class to the selected item
if chat and chat.id in self._chat_items:
self._chat_items[chat.id].add_class("active")
def add_new_chat(self, chat: BasicChat):
"""Add a new chat to the list."""
container = self.query_one("#chats-container", VerticalScroll)
# Remove empty state if present
try:
empty_state = container.query_one(".empty-state")
empty_state.remove()
except LookupError:
# No empty state widget present, which is fine
pass
# Add new chat at the top
chat_item = ChatItem(chat)
self._chat_items[chat.id] = chat_item
# Mount at the beginning
if container.children:
container.mount(chat_item, before=container.children[0])
else:
container.mount(chat_item)
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button press."""
if event.button.id == "new-chat-btn":
self.post_message(NewChatRequested())
Key features:
ChatItem- Individual conversation with title and summary
ChatSelected- Custom message when chat is clicked
NewChatRequested- Message when "New Chat" button is pressed
- Active chat highlighting
- Hover effects
- Error handling for API failures
Step 2: Create the Streaming Handler
Now src/pieces_copilot_tui/streaming_handler.py - connects Pieces OS streaming to our UI:
# src/pieces_copilot_tui/streaming_handler.py
"""Handler for streaming responses from Pieces OS."""
from typing import Callable, Optional
from pieces_os_client.wrapper import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat
class StreamingHandler:
"""Handles streaming responses from the Pieces Copilot."""
def __init__(
self,
pieces_client: PiecesClient,
on_thinking_started: Callable[[], None],
on_stream_started: Callable[[str], None],
on_stream_chunk: Callable[[str], None],
on_stream_completed: Callable[[Optional[BasicChat]], None],
on_stream_error: Callable[[str], None],
):
"""
Initialize the streaming handler.
Args:
pieces_client: The Pieces client instance
on_thinking_started: Callback when thinking starts
on_stream_started: Callback when streaming starts (with initial text)
on_stream_chunk: Callback for each chunk (with full accumulated text)
on_stream_completed: Callback when streaming completes (with optional new chat)
on_stream_error: Callback when an error occurs (with error message)
"""
self.pieces_client = pieces_client
self.on_thinking_started = on_thinking_started
self.on_stream_started = on_stream_started
self.on_stream_chunk = on_stream_chunk
self.on_stream_completed = on_stream_completed
self.on_stream_error = on_stream_error
self._current_response = ""
self._current_status: Optional[str] = None
# Register callback
if self.pieces_client.copilot.ask_stream_ws:
self.pieces_client.copilot.ask_stream_ws.on_message_callback = (
self._handle_stream_message
)
def ask_question(self, query: str) -> None:
"""
Ask a question to the copilot.
Args:
query: The question to ask
"""
self._current_response = ""
self._current_status = None
self.on_thinking_started()
self.pieces_client.copilot.stream_question(query)
def _handle_stream_message(self, response) -> None:
"""Handle streaming messages from the copilot."""
try:
status = response.status
if status == "IN-PROGRESS":
if response.question:
answers = response.question.answers.iterable
for answer in answers:
if answer.text:
if not self._current_response:
# First chunk - start streaming
self._current_response = answer.text
self.on_stream_started(self._current_response)
else:
# Subsequent chunks
self._current_response += answer.text
self.on_stream_chunk(self._current_response)
elif status == "COMPLETED":
# Streaming completed
new_chat = None
if response.conversation:
new_chat = BasicChat(response.conversation)
self.pieces_client.copilot.chat = new_chat
self.on_stream_completed(new_chat)
self._current_response = ""
elif status in ["FAILED", "STOPPED", "CANCELED"]:
# Handle error
error_msg = getattr(response, "error_message", "Unknown error")
self.on_stream_error(error_msg)
self._current_response = ""
except (AttributeError, ConnectionError, ValueError) as e:
self.on_stream_error(str(e))
self._current_response = ""
Step 3: Create the Full View Orchestrator
Now the complete chat_view.py - replaces the simple version from Part 2:
# chat_view.py
"""Main copilot view combining all widgets."""
from typing import Optional
from textual.screen import Screen
from textual.containers import Horizontal, Vertical
from textual.binding import Binding
from textual.widgets import Footer
from pieces_os_client import PiecesClient
from pieces_os_client.wrapper.basic_identifier.chat import BasicChat
from .chat_panel import ChatPanel
from .chats_list import ChatsList, ChatSelected, NewChatRequested
from .chat_input import ChatInput, MessageSubmitted
from .streaming_handler import StreamingHandler
class CopilotView(Screen):
"""Main copilot view screen."""
BINDINGS = [
Binding("ctrl+n", "new_chat", "New Chat"),
Binding("ctrl+r", "rename_chat", "Rename"),
Binding("ctrl+d", "delete_chat", "Delete"),
Binding("ctrl+l", "toggle_ltm", "Toggle LTM"),
Binding("ctrl+q", "quit", "Quit"),
]
DEFAULT_CSS = """
CopilotView {
layout: vertical;
}
CopilotView Horizontal {
height: 1fr;
}
CopilotView Vertical.main-content {
width: 75%;
layout: vertical;
}
"""
def __init__(self, pieces_client: PiecesClient, **kwargs):
super().__init__(**kwargs)
self.pieces_client = pieces_client
self.current_chat: Optional[BasicChat] = None
self.chat_panel: Optional[ChatPanel] = None
self.chats_list: Optional[ChatsList] = None
self.chat_input: Optional[ChatInput] = None
self._ltm_enabled = False
self.streaming_handler: Optional[StreamingHandler] = None
def compose(self):
"""Compose the view."""
with Horizontal():
# Left sidebar - conversations list
self.chats_list = ChatsList()
yield self.chats_list
# Right side - chat panel and input
with Vertical(classes="main-content"):
self.chat_panel = ChatPanel()
yield self.chat_panel
self.chat_input = ChatInput()
yield self.chat_input
yield Footer()
def on_mount(self) -> None:
"""Initialize the view."""
# Load chats
self.chats_list.load_chats(self.pieces_client)
# Show welcome message
self.chat_panel.show_welcome()
# Check LTM status
try:
self._ltm_enabled = self.pieces_client.copilot.is_chat_ltm_enabled()
except (ConnectionError, AttributeError):
self._ltm_enabled = False
# Setup streaming handler
self.streaming_handler = StreamingHandler(
pieces_client=self.pieces_client,
on_thinking_started=self._on_thinking_started,
on_stream_started=self._on_stream_started,
on_stream_chunk=self._on_stream_chunk,
on_stream_completed=self._on_stream_completed,
on_stream_error=self._on_stream_error,
)
def on_chat_selected(self, message: ChatSelected) -> None:
"""Handle chat selection."""
self.current_chat = message.chat
self.pieces_client.copilot.chat = message.chat
self.chats_list.set_active_chat(message.chat)
# Load conversation
self.chat_panel.clear_messages()
self.chat_panel.border_title = f"Chat: {message.chat.name}"
try:
messages = message.chat.messages()
for msg in messages:
role = msg.role
content = msg.raw_content
self.chat_panel.add_message(role, content)
except (ConnectionError, AttributeError, ValueError) as e:
self.chat_panel.add_message("system", f"❌ Error loading messages: {str(e)}")
def on_new_chat_requested(self, _: NewChatRequested) -> None:
"""Handle new chat request."""
self.action_new_chat()
def on_message_submitted(self, message: MessageSubmitted) -> None:
"""Handle user message submission."""
if self.chat_panel.is_streaming_active():
return
# Add user message to chat panel
self.chat_panel.add_message("user", message.text)
# Send to copilot via streaming handler
if self.streaming_handler:
self.streaming_handler.ask_question(message.text)
# Streaming handler callbacks
def _on_thinking_started(self) -> None:
"""Called when copilot starts thinking."""
self.call_from_thread(self.chat_panel.add_thinking_indicator)
def _on_stream_started(self, initial_text: str) -> None:
"""Called when streaming starts with initial text."""
self.call_from_thread(
self.chat_panel.add_streaming_message,
"assistant",
initial_text
)
def _on_stream_chunk(self, full_text: str) -> None:
"""Called for each streaming chunk with accumulated text."""
self.call_from_thread(
self.chat_panel.update_streaming_message,
full_text
)
def _on_stream_completed(self, new_chat: Optional[BasicChat]) -> None:
"""Called when streaming completes."""
self.call_from_thread(self.chat_panel.finalize_streaming_message)
# Update chat reference if new chat was created
if new_chat:
old_chat = self.current_chat
if not old_chat or old_chat.id != new_chat.id:
self.current_chat = new_chat
self.call_from_thread(self.chats_list.add_new_chat, new_chat)
self.call_from_thread(self.chats_list.set_active_chat, new_chat)
self.chat_panel.border_title = f"Chat: {new_chat.name}"
def _on_stream_error(self, error_msg: str) -> None:
"""Called when a streaming error occurs."""
self.call_from_thread(
self.chat_panel.add_message,
"system",
f"❌ Error: {error_msg}"
)
def action_new_chat(self):
"""Create a new chat."""
self.current_chat = None
self.pieces_client.copilot.chat = None
self.chats_list.set_active_chat(None)
self.chat_panel.clear_messages()
self.chat_panel.border_title = "Chat: New Conversation"
self.chat_panel.show_welcome()
self.chat_input.focus()
def action_rename_chat(self):
"""Rename the current chat."""
if not self.current_chat:
self.notify("No chat selected", severity="warning")
return
# For now, just show a notification - you can implement a dialog later
self.notify("Rename chat - Not implemented yet", severity="info")
def action_delete_chat(self):
"""Delete the current chat."""
if not self.current_chat:
self.notify("No chat selected", severity="warning")
return
try:
chat_name = self.current_chat.name
self.current_chat.delete()
self.notify(f"Deleted chat: {chat_name}", severity="success")
# Clear view and reload chats
self.action_new_chat()
self.chats_list.load_chats(self.pieces_client)
except (ConnectionError, AttributeError) as e:
self.notify(f"Error deleting chat: {str(e)}", severity="error")
def action_toggle_ltm(self):
"""Toggle LTM (Long Term Memory)."""
try:
is_chat_ltm_enabled = self.pieces_client.copilot.is_chat_ltm_enabled()
if is_chat_ltm_enabled:
self.pieces_client.copilot.deactivate_ltm()
self.notify("🧠 Chat LTM disabled", severity="info")
self._ltm_enabled = False
else:
is_system_ltm_running = self.pieces_client.copilot.is_ltm_running()
if is_system_ltm_running:
self.pieces_client.copilot.activate_ltm()
self.notify("🧠 Chat LTM enabled", severity="success")
self._ltm_enabled = True
else:
self.notify("🧠 LTM system not running. Please enable it first.", severity="warning")
except (ConnectionError, AttributeError) as e:
self.notify(f"❌ Error toggling LTM: {str(e)}", severity="error")
🚨 Critical for Thread Safety
Always use call_from_thread() when updating UI from background threads. Direct UI updates from threads will cause crashes! The streaming handler runs in a background thread, so all UI updates must go through this method.
Key features:
- Split-pane layout (25% sidebar, 75% chat)
- Handles all custom messages (
ChatSelected,NewChatRequested,MessageSubmitted)
- Thread-safe UI updates via
call_from_thread()
- Keyboard shortcuts (Ctrl+N, Ctrl+D, Ctrl+L, Ctrl+Q)
- Conversation management
- LTM toggle support
Step 4: Update the Main Application
Now update src/pieces_copilot_tui/app.py to use the full CopilotView:
# src/pieces_copilot_tui/app.py
"""Pieces Copilot TUI - A Terminal User Interface for Pieces OS."""
from textual.app import App
from pieces_os_client.wrapper import PiecesClient
from .chat_view import CopilotView
class PiecesCopilotTUI(App):
"""Pieces Copilot TUI with full features."""
# CSS styles for the entire app
DEFAULT_CSS = """
Screen {
background: $background;
color: $text;
}
.error {
color: $error;
text-style: bold;
}
.success {
color: $success;
text-style: bold;
}
.warning {
color: $warning;
text-style: bold;
}
"""
def __init__(self, pieces_client: PiecesClient = None, **kwargs):
super().__init__(**kwargs)
self.pieces_client = pieces_client or PiecesClient()
self.copilot_view = None
def on_mount(self) -> None:
"""Initialize the application."""
self.title = "Pieces Copilot TUI"
# Initialize Pieces client
if not self.pieces_client.connect_websocket():
self.notify("Failed to connect to Pieces OS", severity="error")
self.exit()
return
# Create and push the full copilot view
self.copilot_view = CopilotView(pieces_client=self.pieces_client)
self.push_screen(self.copilot_view)
def run_tui():
"""Run the Pieces Copilot TUI application."""
# Initialize Pieces client
pieces_client = PiecesClient()
# Check if Pieces OS is running
if not pieces_client.is_pieces_running():
print("Error: Pieces OS is not running. Please start Pieces OS first.")
return
app = PiecesCopilotTUI(pieces_client=pieces_client)
app.run()
if __name__ == "__main__":
run_tui()
Changes from Part 2:
- Import
CopilotViewinstead ofSimpleChatView
- Added
warningCSS class
- Now using the full-featured view
Step 5: Run Your Complete TUI
The __main__.py file is already set up from Part 2. Run it the Pythonic way:
# Navigate to src directory
cd src
# Run the module
python -m pieces_copilot_tui
That's it! Simple, clean, and Pythonic. ✨
🎉 You now have a fully functional Pieces Copilot TUI!
Add Debug Logging
# In any widget method
self.log(f"Current state: {some_variable}")
self.log.info("Info message")
self.log.warning("Warning message")
self.log.error("Error message")
# View in textual console
Common Issues
Problem: Streaming doesn't start
# Check WebSocket connection
if not client.copilot.ask_stream_ws:
print("❌ WebSocket not connected!")
client.connect_websocket()
Problem: UI not updating from streaming
# Make sure you're using call_from_thread
self.call_from_thread(self.chat_panel.add_message, "assistant", text)
# NOT: self.chat_panel.add_message("assistant", text)
Problem: Messages not loading
# Add error handling
try:
messages = chat.messages()
except Exception as e:
self.log.error(f"Failed to load messages: {e}")