Architecture

Consoul uses a three-layer architecture that cleanly separates business logic from presentation, enabling headless SDK usage for backend integrations while supporting rich TUI and CLI experiences.

Overview

The architecture consists of three distinct layers:

graph TB
    subgraph "User Interfaces"
        TUI[TUI Layer<br/>Rich Terminal UI<br/>Textual Widgets]
        CLI[CLI Layer<br/>Command Line<br/>Typer Commands]
    end

    subgraph "SDK Layer (Headless)"
        CS[ConversationService]
        TS[ToolService]
        MS[ModelService]
    end

    subgraph "Core Components"
        LLM[LangChain Models]
        TR[Tool Registry]
        CH[Conversation History]
        CFG[Configuration]
    end

    TUI --> CS
    TUI --> TS
    TUI --> MS
    CLI --> CS
    CLI --> TS
    CLI --> MS

    CS --> LLM
    CS --> TR
    CS --> CH
    TS --> TR
    MS --> LLM
    MS --> CFG

    style TUI fill:#e1f5ff
    style CLI fill:#e1f5ff
    style CS fill:#fff3e0
    style TS fill:#fff3e0
    style MS fill:#fff3e0

Layer Responsibilities

SDK Layer

Location: src/consoul/sdk/, src/consoul/ai/

Purpose: Headless business logic with no UI dependencies

Components:

ConversationService - AI conversation management with streaming
ToolService - Tool configuration and approval management
ModelService - Model initialization and switching
Protocols - Callback interfaces (e.g., ToolExecutionCallback)
Models - Data classes (e.g., Token, ToolRequest, ConversationStats)

Key Characteristics:

✅ No imports from consoul.tui or consoul.cli
✅ Async-first API design
✅ Protocol-based callbacks for UI integration
✅ Comprehensive docstrings with examples
✅ Unit testable with mocked dependencies

Example: src/consoul/sdk/services/conversation.py:42

class ConversationService:
    """Service layer for AI conversation management.

    Provides headless conversation interface with streaming responses,
    tool execution approval, multimodal message support, and cost tracking.
    Completely decoupled from TUI/CLI to enable SDK-first architecture.
    """

    async def send_message(
        self,
        content: str,
        *,
        attachments: list[Attachment] | None = None,
        on_tool_request: ToolApprovalCallback | None = None,
    ) -> AsyncIterator[Token]:
        """Send message and stream AI response.

        Args:
            content: User message text
            attachments: Optional file attachments
            on_tool_request: Optional callback for tool execution approval

        Yields:
            Token: Streaming tokens with content, cost, and metadata
        """
        # Business logic here, no UI code

TUI Layer

Location: src/consoul/tui/

Purpose: Rich terminal user interface using Textual framework

Components:

ConsoulApp - Main Textual application
Widgets - Custom UI components (chat display, input field, modals)
Screens - Full-screen UI layouts
Themes - Visual styling and color schemes

Key Characteristics:

✅ Imports SDK services for business logic
✅ Handles user input and presentation only
✅ Uses callback protocols to provide approval UI
✅ Integration tested with Textual test harness

Example: TUI widget using ConversationService

from consoul.sdk.services import ConversationService
from consoul.sdk.models import ToolRequest

class ChatWidget(Widget):
    def __init__(self):
        self.service = ConversationService.from_config()

    async def send_user_message(self, message: str):
        """Send message and display streaming response."""
        # UI provides approval callback
        async for token in self.service.send_message(
            message,
            on_tool_request=self.approval_provider  # UI callback
        ):
            self.display_token(token.content)  # Presentation only

CLI Layer

Location: src/consoul/cli/

Purpose: Command-line interface using Typer framework

Components:

Commands - CLI command handlers
Arguments - Argument parsing and validation
Output - Terminal output formatting

Key Characteristics:

✅ Imports SDK services for business logic
✅ Handles argument parsing and simple output
✅ Non-interactive approval (auto-approve or deny)

Example: CLI command using ConversationService

from consoul.sdk.services import ConversationService
import typer

app = typer.Typer()

@app.command()
def chat(message: str):
    """Send a chat message."""
    service = ConversationService.from_config()

    async for token in service.send_message(message):
        print(token.content, end="", flush=True)  # Output only

Architecture Boundaries

Import Rules

SDK Layer (src/consoul/sdk/, src/consoul/ai/):

✅ Can import: langchain, pydantic, standard library
❌ Cannot import: consoul.tui, consoul.cli, textual, typer, rich

TUI Layer (src/consoul/tui/):

✅ Can import: consoul.sdk, consoul.ai, textual, rich
❌ Cannot import: consoul.cli

CLI Layer (src/consoul.cli/):

✅ Can import: consoul.sdk, consoul.ai, typer
❌ Cannot import: consoul.tui, textual, rich

Where Code Belongs

Feature	SDK Layer	TUI Layer	CLI Layer
AI conversation logic	✅ ConversationService	❌	❌
Token streaming	✅ AsyncIterator[Token]	❌	❌
Tool execution	✅ ToolService	❌	❌
Model switching	✅ ModelService	❌	❌
User approval UI	❌	✅ Modal dialog	❌
Message display	❌	✅ Chat widget	✅ print()
Command parsing	❌	❌	✅ Typer
Configuration loading	✅ load_config()	Calls SDK	Calls SDK

Data Flow

Message Flow (TUI Example)

sequenceDiagram
    participant User
    participant TUI as TUI Widget
    participant CS as ConversationService
    participant Model as LangChain Model
    participant TR as Tool Registry

    User->>TUI: Type message
    TUI->>CS: send_message(text, on_tool_request=callback)
    CS->>Model: stream(messages)

    loop For each token
        Model-->>CS: Token chunk
        CS-->>TUI: yield Token
        TUI-->>User: Display token
    end

    Model-->>CS: Tool call detected
    CS->>TR: Execute tool
    TR->>TUI: on_tool_request(request)
    TUI->>User: Show approval modal
    User->>TUI: Approve/Deny
    TUI-->>TR: approved=True/False
    TR-->>CS: Tool result
    CS->>Model: Continue with result

Backend Integration Flow (FastAPI Example)

sequenceDiagram
    participant WS as WebSocket Client
    participant Server as FastAPI Server
    participant CS as ConversationService
    participant Model as LangChain Model

    WS->>Server: Connect
    Server->>CS: ConversationService.from_config()

    WS->>Server: {type: "message", content: "..."}
    Server->>CS: send_message(content, on_tool_request=ws_callback)

    loop Streaming
        CS-->>Server: yield Token
        Server-->>WS: {type: "token", content: "..."}
    end

    CS-->>Server: Done
    Server-->>WS: {type: "done"}

Good vs Bad Examples

❌ Bad: Business Logic in UI

class ChatWidget(Widget):
    async def send_message(self, msg: str):
        # BAD: AI logic in UI layer
        from langchain_openai import ChatOpenAI
        model = ChatOpenAI(model="gpt-4o")

        # BAD: Direct model interaction
        response = await model.ainvoke([{"role": "user", "content": msg}])

        # BAD: Tool execution in UI
        if response.tool_calls:
            for tool_call in response.tool_calls:
                result = self.execute_tool(tool_call)  # UI shouldn't do this

        self.display(response.content)

Problems:

Business logic coupled to UI
Cannot be used headlessly
Hard to unit test
Code duplication with CLI

✅ Good: Service Handles Logic

class ChatWidget(Widget):
    def __init__(self):
        # GOOD: Use SDK service
        self.service = ConversationService.from_config()

    async def send_message(self, msg: str):
        # GOOD: Service handles all AI logic
        async for token in self.service.send_message(
            msg,
            on_tool_request=self.approval_provider  # UI provides callback
        ):
            # GOOD: UI only displays
            self.display_token(token.content)

Benefits:

Business logic in SDK layer
Same service used in TUI, CLI, FastAPI, etc.
Easy to unit test services
UI code is simple and focused

❌ Bad: SDK Importing UI

# src/consoul/sdk/services/conversation.py

from rich.console import Console  # BAD: UI dependency in SDK

class ConversationService:
    def __init__(self):
        self.console = Console()  # BAD: Couples to Rich

    async def send_message(self, msg: str):
        self.console.print("[bold]Sending...[/bold]")  # BAD: UI in service
        # ...

Problems:

SDK cannot be used in FastAPI/WebSocket (Rich requires terminal)
Breaks headless usage
Forces UI choice on SDK consumers

✅ Good: Protocol-Based Callbacks

# src/consoul/sdk/protocols.py

class ToolExecutionCallback(Protocol):
    """Protocol for tool execution approval."""
    async def on_tool_request(self, request: ToolRequest) -> bool:
        """Called when tool execution needs approval."""
        ...

# src/consoul/sdk/services/conversation.py

class ConversationService:
    async def send_message(
        self,
        msg: str,
        on_tool_request: ToolExecutionCallback | None = None
    ):
        # Service calls protocol method, UI provides implementation
        if on_tool_request:
            approved = await on_tool_request.on_tool_request(request)

Benefits:

SDK defines contract (protocol)
UI provides implementation (modal, WebSocket, CLI, etc.)
Loose coupling via protocol
Supports any approval mechanism

Integration Patterns

FastAPI WebSocket Integration

See examples/fastapi_websocket_server.py for full example:

from fastapi import FastAPI, WebSocket
from consoul.sdk.services import ConversationService

app = FastAPI()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()

    # Create service per connection
    service = ConversationService.from_config()

    # Stream responses
    async for token in service.send_message(user_message):
        await websocket.send_json({"token": token.content})

Custom Approval Provider

Implement the ToolExecutionCallback protocol for custom approval logic:

from consoul.sdk.protocols import ToolExecutionCallback
from consoul.sdk.models import ToolRequest

class WebSocketApprovalProvider:
    def __init__(self, websocket):
        self.websocket = websocket

    async def on_tool_request(self, request: ToolRequest) -> bool:
        # Send approval request to WebSocket client
        await self.websocket.send_json({
            "type": "tool_request",
            "id": request.id,
            "name": request.name,
            "risk_level": request.risk_level
        })

        # Wait for client response
        response = await self.websocket.receive_json()
        return response.get("approved", False)

# Use in service
async for token in service.send_message(
    msg,
    on_tool_request=WebSocketApprovalProvider(websocket)
):
    # ...

Testing Strategy

See testing.md for detailed testing guidelines.

SDK Layer Testing

Unit tests with mocked dependencies:

import pytest
from unittest.mock import AsyncMock, Mock
from consoul.sdk.services import ConversationService

@pytest.mark.asyncio
async def test_conversation_service_streaming():
    # Mock LangChain model
    mock_model = Mock()
    mock_model.stream = Mock(return_value=[
        Mock(content="Hello"),
        Mock(content=" world")
    ])

    # Create service with mocked model
    service = ConversationService(
        model=mock_model,
        conversation=mock_conversation,
        tool_registry=None
    )

    # Test streaming
    tokens = []
    async for token in service.send_message("Hi"):
        tokens.append(token.content)

    assert "".join(tokens) == "Hello world"
    mock_model.stream.assert_called_once()

TUI Layer Testing

Integration tests with Textual test harness:

from textual.widgets import Input
from consoul.tui.app import ConsoulApp

async def test_tui_message_send():
    app = ConsoulApp()
    async with app.run_test() as pilot:
        # Type message
        await pilot.press("i")  # Enter input mode
        await pilot.press(*"Hello AI")
        await pilot.press("enter")

        # Verify message displayed
        assert "Hello AI" in app.query_one(ChatDisplay).text

Common Pitfalls

❌ Pitfall: Importing Rich in SDK

Problem: from rich.console import Console in SDK layer

Why it's bad: Rich requires a terminal, breaks headless usage

Solution: Use protocols and let UI provide display logic

❌ Pitfall: Business Logic in TUI

Problem: Direct LangChain calls in TUI widgets

Why it's bad: Logic cannot be reused in CLI or backends

Solution: Move logic to ConversationService, call from TUI

❌ Pitfall: Blocking Calls in Services

Problem: Using synchronous I/O in service methods

Why it's bad: Blocks event loop in TUI and async servers

Solution: Use async methods and asyncio.run_in_executor() for sync code

Next Steps

Service Layer Guide - Learn when and how to extend services
Testing Guide - Comprehensive testing patterns for each layer
API Reference - Service layer API documentation
SDK Integration Examples - Real-world integration patterns

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Architecture

Overview

Layer Responsibilities

SDK Layer

TUI Layer

CLI Layer

Architecture Boundaries

Import Rules

Where Code Belongs

Data Flow

Message Flow (TUI Example)

Backend Integration Flow (FastAPI Example)

Good vs Bad Examples

❌ Bad: Business Logic in UI

✅ Good: Service Handles Logic

❌ Bad: SDK Importing UI

✅ Good: Protocol-Based Callbacks

Integration Patterns

FastAPI WebSocket Integration

Custom Approval Provider

Testing Strategy

SDK Layer Testing

TUI Layer Testing

Common Pitfalls

❌ Pitfall: Importing Rich in SDK

❌ Pitfall: Business Logic in TUI

❌ Pitfall: Blocking Calls in Services

Next Steps

FilesExpand file tree

architecture.md

Latest commit

History

architecture.md

File metadata and controls

Architecture

Overview

Layer Responsibilities

SDK Layer

TUI Layer

CLI Layer

Architecture Boundaries

Import Rules

Where Code Belongs

Data Flow

Message Flow (TUI Example)

Backend Integration Flow (FastAPI Example)

Good vs Bad Examples

❌ Bad: Business Logic in UI

✅ Good: Service Handles Logic

❌ Bad: SDK Importing UI

✅ Good: Protocol-Based Callbacks

Integration Patterns

FastAPI WebSocket Integration

Custom Approval Provider

Testing Strategy

SDK Layer Testing

TUI Layer Testing

Common Pitfalls

❌ Pitfall: Importing Rich in SDK

❌ Pitfall: Business Logic in TUI

❌ Pitfall: Blocking Calls in Services

Next Steps