🚀 Technical Briefing: This tutorial is part of our deep-dive series on Agentic Workflows at Gate of AI. For the full technical breakdown, interactive code sandbox, and the native Arabic translation, visit the original article here.
<span>Tutorial</span>
<span>Intermediate</span>
<span>48 min read</span>
<span>© Gate of AI 2026-08-21</span>
<p>Build a Python terminal assistant with two explicit model providers, SQLite conversation sessions, bounded history, controlled retries, and tests.</p>
<h2>What You Will Build</h2>
<p>This tutorial builds a local Python chat application that lets a user choose between an OpenAI ChatGPT-style model and Anthropic Claude Sonnet 5. The application uses one internal message format, stores sessions in SQLite, keeps only a bounded number of recent messages in each request, and makes provider selection visible in the terminal.</p>
<p>The timing matters. Anthropic introduced Claude Sonnet 5 on June 30, 2026 as its most agentic Sonnet model. Anthropic says the model can make plans, use tools such as browsers and terminals, and run autonomously at a capability level that recently required larger and more expensive models. It also positions Sonnet 5 as close to Opus 4.8 performance at lower prices, with improvements over Sonnet 4.6 in reasoning, tool use, coding, and knowledge work.</p>
<p>That does not mean a local chat client should automatically give a model access to a browser, terminal, customer system, or internal database. This tutorial deliberately implements text chat only. It creates a dependable boundary for model comparison and conversational workflows first. If you later add tools, deterministic application code should validate permissions, arguments, timeouts, and approval requirements before any external action is executed.</p>
<p>This approach is useful for engineering teams in the GCC and Middle East that need to evaluate more than one AI provider while retaining control over their application architecture. The application does not silently send a failed Claude request to OpenAI, or the reverse. The user chooses the provider, which makes routing behaviour visible during technical evaluation and governance review.</p>
<h3>Architecture</h3>
<ul>
<li><code>config.py</code> reads required environment variables and validates safe local limits.</li>
<li><code>providers.py</code> converts one internal conversation format into each provider's request format.</li>
<li><code>storage.py</code> creates durable SQLite sessions and retrieves chronological recent history.</li>
<li><code>chat.py</code> provides the terminal loop, commands, controlled retries, and provider routing.</li>
</ul>
<p>The provider adapter is the important design decision. The rest of the program depends on a small internal contract rather than directly on a vendor SDK. That makes the application easier to test and lets you add an approved internal gateway later without rewriting persistence or command handling.</p>
<h2>Prerequisites and Setup</h2>
<ul>
<li>Python 3.10 or newer.</li>
<li>An OpenAI API key and a model identifier available to your account.</li>
<li>An Anthropic API key and access to Claude Sonnet 5.</li>
<li>Basic familiarity with virtual environments, environment variables, and the terminal.</li>
<li>SQLite, which is included with standard CPython installations.</li>
</ul>
<pre><code>mkdir multi-model-chat
cd multi-model-chat
python -m venv .venv
macOS and Linux
source .venv/bin/activate
Windows PowerShell
..venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install openai anthropic rich pytest
mkdir tests
<p>Do not place credentials in Python source files. Set them in your shell, CI secret store, container runtime, or approved deployment secret manager. The application requires a configured model name for each provider because model availability is account-specific.</p>
<pre><code># macOS and Linux
export OPENAI_API_KEY="your-openai-key"
export OPENAI_MODEL="your-openai-model"
export ANTHROPIC_API_KEY="your-anthropic-key"
export ANTHROPIC_MODEL="claude-sonnet-5"
Optional local limits
export MAX_HISTORY_MESSAGES="20"
export MAX_OUTPUT_TOKENS="1200"
export REQUEST_TIMEOUT_SECONDS="60"
<p>On Windows PowerShell, use <code>$env:OPENAI_API_KEY="..."</code> syntax instead. In a production deployment, inject these same variable names through the platform's managed secret mechanism. Never print keys in logs, commit them to Git, or ship them to browser code.</p>
<h2>Step 1: Add Configuration Validation</h2>
<p>Create <code>config.py</code>. This small module keeps configuration out of business logic and fails early when a limit is invalid. It does not require a dotenv dependency; environment variables are its only input.</p>
<pre><code>from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class Settings:
openai_api_key: str | None
openai_model: str | None
anthropic_api_key: str | None
anthropic_model: str | None
max_history_messages: int
max_output_tokens: int
request_timeout_seconds: float
database_path: Path
def require_openai(self) -> tuple[str, str]:
if not self.openai_api_key or not self.openai_model:
raise RuntimeError(
"OPENAI_API_KEY and OPENAI_MODEL are required for OpenAI."
)
return self.openai_api_key, self.openai_model
def require_anthropic(self) -> tuple[str, str]:
if not self.anthropic_api_key or not self.anthropic_model:
raise RuntimeError(
"ANTHROPIC_API_KEY and ANTHROPIC_MODEL are required for Anthropic."
)
return self.anthropic_api_key, self.anthropic_model
def read_positive_int(name: str, default: int, minimum: int) -> int:
value = int(os.getenv(name, str(default)))
if value < minimum:
raise ValueError(f"{name} must be at least {minimum}.")
return value
def get_settings() -> Settings:
timeout = float(os.getenv("REQUEST_TIMEOUT_SECONDS", "60"))
if timeout <= 0:
raise ValueError("REQUEST_TIMEOUT_SECONDS must be greater than zero.")
return Settings(
openai_api_key=os.getenv("OPENAI_API_KEY"),
openai_model=os.getenv("OPENAI_MODEL"),
anthropic_api_key=os.getenv("ANTHROPIC_API_KEY"),
anthropic_model=os.getenv("ANTHROPIC_MODEL"),
max_history_messages=read_positive_int(
"MAX_HISTORY_MESSAGES", default=20, minimum=2
),
max_output_tokens=read_positive_int(
"MAX_OUTPUT_TOKENS", default=1200, minimum=1
),
request_timeout_seconds=timeout,
database_path=Path(os.getenv("SQLITE_DATABASE_PATH", "chat_history.sqlite3")),
)</code></pre>
<p>A message-count limit is a simple safeguard, not a token counter. Different models can tokenize the same text differently, and a short character count is not a reliable proxy for request size. Keeping the trimming rule isolated means you can replace it later with provider-aware token budgeting or summarisation.</p>
<h2>Step 2: Create the Provider Adapter Layer</h2>
<p>Create <code>providers.py</code>. OpenAI and Anthropic use different request and response shapes. The adapter converts both responses into <code>CompletionResult</code>, so the CLI does not need provider-specific parsing code.</p>
<pre><code>from __future__ import annotations
from dataclasses import dataclass
from typing import Literal, Protocol, Sequence
from anthropic import Anthropic
from openai import OpenAI
from config import Settings
Role = Literal["user", "assistant"]
ProviderName = Literal["openai", "anthropic"]
@dataclass(frozen=True)
class ChatMessage:
role: Role
content: str
@dataclass(frozen=True)
class CompletionResult:
provider: ProviderName
model: str
text: str
input_tokens: int | None
output_tokens: int | None
class ChatProvider(Protocol):
name: ProviderName
def complete(
self,
system_prompt: str,
messages: Sequence[ChatMessage],
max_output_tokens: int,
) -> CompletionResult:
...
class OpenAIChatProvider:
name: ProviderName = "openai"
def __init__(self, settings: Settings) -> None:
api_key, model = settings.require_openai()
self._model = model
self._client = OpenAI(
api_key=api_key,
timeout=settings.request_timeout_seconds,
max_retries=0,
)
def complete(
self,
system_prompt: str,
messages: Sequence[ChatMessage],
max_output_tokens: int,
) -> CompletionResult:
response = self._client.chat.completions.create(
model=self._model,
messages=[
{"role": "system", "content": system_prompt},
*[{"role": message.role, "content": message.content} for message in messages],
],
max_tokens=max_output_tokens,
)
text = (response.choices[0].message.content or "").strip()
if not text:
raise RuntimeError("OpenAI returned an empty assistant response.")
usage = response.usage
return CompletionResult(
provider=self.name,
model=response.model,
text=text,
input_tokens=usage.prompt_tokens if usage else None,
output_tokens=usage.completion_tokens if usage else None,
)
class AnthropicChatProvider:
name: ProviderName = "anthropic"
def __init__(self, settings: Settings) -> None:
api_key, model = settings.require_anthropic()
self._model = model
self._client = Anthropic(
api_key=api_key,
timeout=settings.request_timeout_seconds,
max_retries=0,
)
def complete(
self,
system_prompt: str,
messages: Sequence[ChatMessage],
max_output_tokens: int,
) -> CompletionResult:
response = self._client.messages.create(
model=self._model,
system=system_prompt,
messages=[
{"role": message.role, "content": message.content}
for message in messages
],
max_tokens=max_output_tokens,
)
text = "\n".join(
block.text
for block in response.content
if getattr(block, "type", None) == "text"
).strip()
if not text:
raise RuntimeError("Anthropic returned no text content.")
usage = response.usage
return CompletionResult(
provider=self.name,
model=response.model,
text=text,
input_tokens=usage.input_tokens if usage else None,
output_tokens=usage.output_tokens if usage else None,
)
def create_provider(name: ProviderName, settings: Settings) -> ChatProvider:
if name == "openai":
return OpenAIChatProvider(settings)
return AnthropicChatProvider(settings)
<p>The OpenAI client uses the modern object-oriented SDK pattern: <code>from openai import OpenAI</code>, then <code>client.chat.completions.create()</code>. The application disables SDK retries so one application-level retry policy remains responsible for retry decisions.</p>
<h2>Step 3: Persist Sessions in SQLite</h2>
<p>Create <code>storage.py</code>. SQLite is suitable for this local single-user terminal tool because it provides a durable local database without a separate server. The retrieval query first selects the newest rows, then reorders that selected subset chronologically before sending it to a model.</p>
<pre><code>from __future__ import annotations
import sqlite3
from pathlib import Path
from uuid import uuid4
from providers import ChatMessage, Role
class ChatStore:
def init(self, path: Path) -> None:
self.connection = sqlite3.connect(path)
self.connection.execute("PRAGMA foreign_keys = ON")
self.connection.executescript(
"""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN ('user', 'assistant')),
provider TEXT,
content TEXT NOT NULL,
FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS message_session_order
ON messages(session_id, id);
"""
)
self.connection.commit()
def create_session(self) -> str:
session_id = str(uuid4())
self.connection.execute("INSERT INTO sessions(id) VALUES (?)", (session_id,))
self.connection.commit()
return session_id
def exists(self, session_id: str) -> bool:
return self.connection.execute(
"SELECT 1 FROM sessions WHERE id = ?", (session_id,)
).fetchone() is not None
def add(self, session_id: str, role: Role, content: str, provider: str | None = None) -> None:
self.connection.execute(
"INSERT INTO messages(session_id, role, provider, content) VALUES (?, ?, ?, ?)",
(session_id, role, provider, content),
)
self.connection.commit()
def recent(self, session_id: str, limit: int) -> list[ChatMessage]:
rows = self.connection.execute(
"""
SELECT role, content FROM (
SELECT id, role, content FROM messages
WHERE session_id = ? ORDER BY id DESC LIMIT ?
) ORDER BY id ASC
""",
(session_id, limit),
).fetchall()
return [ChatMessage(role=row[0], content=row[1]) for row in rows]
def clear(self, session_id: str) -> None:
self.connection.execute("DELETE FROM messages WHERE session_id = ?", (session_id,))
self.connection.commit()
def close(self) -> None:
self.connection.close()</code></pre>
<h2>Step 4: Build the Interactive CLI</h2>
<p>Create <code>chat.py</code>. The retry function only retries errors that look like a connection failure, timeout, or server-side error. It does not retry every exception. A missing key, invalid configuration, or rejected request needs correction rather than repeated network traffic.</p>
<pre><code>from __future__ import annotations
import argparse
import time
from typing import Literal
from rich.console import Console
from rich.markdown import Markdown
from config import get_settings
from providers import ProviderName, create_provider
from storage import ChatStore
console = Console()
SYSTEM_PROMPT = "You are a precise and practical technical assistant. State important assumptions."
def retryable(error: Exception) -> bool:
name = type(error).name.lower()
status = getattr(error, "status_code", None)
return "timeout" in name or "connection" in name or (isinstance(status, int) and status >= 500)
def complete_with_retry(provider_name: ProviderName, store: ChatStore, session_id: str) -> object:
settings = get_settings()
provider = create_provider(provider_name, settings)
messages = store.recent(session_id, settings.max_history_messages)
last_error: Exception | None = None
for attempt in range(1, 4):
try:
return provider.complete(SYSTEM_PROMPT, messages, settings.max_output_tokens)
except Exception as error:
last_error = error
if not retryable(error) or attempt == 3:
raise
time.sleep(min(2 ** (attempt - 1), 4))
assert last_error is not None
raise last_error
def arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Local multi-model terminal chat")
parser.add_argument("--provider", choices=["openai", "anthropic"], default="openai")
parser.add_argument("--session")
return parser.parse_args()
def main() -> None:
args = arguments()
settings = get_settings()
store = ChatStore(settings.database_path)
provider_name: ProviderName = args.provider
session_id = args.session or store.create_session()
if args.session and not store.exists(session_id):
raise SystemExit(f"Session does not exist: {session_id}")
console.print(f"[green]Ready.[/green] provider={provider_name} session={session_id}")
console.print("Commands: /provider openai, /provider anthropic, /new, /clear, /status, /exit")
try:
while True:
try:
prompt = console.input("[bold blue]You > [/bold blue]").strip()
except (EOFError, KeyboardInterrupt):
console.print("\n[yellow]Goodbye.[/yellow]")
break
if not prompt:
continue
if prompt == "/exit":
break
if prompt == "/new":
session_id = store.create_session()
console.print(f"[green]New session:[/green] {session_id}")
continue
if prompt == "/clear":
store.clear(session_id)
console.print("[yellow]Current history cleared.[/yellow]")
continue
if prompt == "/status":
console.print(f"provider={provider_name} session={session_id}")
continue
if prompt.startswith("/provider "):
choice = prompt.removeprefix("/provider ").strip().lower()
if choice in {"openai", "anthropic"}:
provider_name = choice
console.print(f"[green]Provider changed to {choice}.[/green]")
else:
console.print("[red]Choose openai or anthropic.[/red]")
continue
if prompt.startswith("/"):
console.print("[red]Unknown command.[/red]")
continue
store.add(session_id, "user", prompt)
try:
result = complete_with_retry(provider_name, store, session_id)
store.add(session_id, "assistant", result.text, result.provider)
console.print(Markdown(result.text))
console.print(
f"[dim]provider={result.provider} model={result.model} "
f"input_tokens={result.input_tokens} output_tokens={result.output_tokens}[/dim]"
)
except Exception as error:
console.print(f"[red]Request failed:[/red] {type(error).__name__}")
console.print("Your user message remains in the local session. Correct configuration or try again.")
finally:
store.close()
if name == "main":
main()
<p>Run the application with <code>python chat.py --provider anthropic</code> or <code>python chat.py --provider openai</code>. Use <code>/provider anthropic</code> during a session to switch explicitly. The previous retained conversation remains available because both adapters consume the same internal message structure.</p>
<p>Saving the user message before the request means an interrupted or failed request remains visible in local history. That is useful for a local tool, but it can leave a user turn without an assistant reply. A larger deployment can add delivery states such as pending, completed, and failed.</p>
<h2>Test the Storage Layer</h2>
<p>Create <code>tests/test_storage.py</code>. These tests use a temporary SQLite database and do not require API keys or provider network calls.</p>
<pre><code>from storage import ChatStore
def test_recent_messages_are_chronological(tmp_path):
store = ChatStore(tmp_path / "test.sqlite3")
session = store.create_session()
store.add(session, "user", "one")
store.add(session, "assistant", "two", "openai")
store.add(session, "user", "three")
messages = store.recent(session, 2)
assert [(message.role, message.content) for message in messages] == [
("assistant", "two"),
("user", "three"),
]
store.close()
def test_clear_preserves_session(tmp_path):
store = ChatStore(tmp_path / "test.sqlite3")
session = store.create_session()
store.add(session, "user", "remove")
store.clear(session)
assert store.exists(session)
assert store.recent(session, 20) == []
store.close()</code></pre>
<pre><code>python -m pytest -q
python -m py_compile config.py providers.py storage.py chat.py
python chat.py --provider anthropic
<h3>Production Boundaries Before You Add Agents</h3>
<p>Claude Sonnet 5 is designed for more agentic work, including planning and tool use. Treat that capability as a reason to strengthen your application boundary, not weaken it. Keep model-generated suggestions separate from execution. Use allowlisted tools, typed inputs, short timeouts, identity and authorisation checks, audit records, and approval steps for consequential operations.</p>
<p>For a multi-user web application, replace the local terminal interface with an authenticated backend, move from local SQLite to a database suited to the deployment's concurrency needs, and keep provider API keys on the server. Add structured operational logs that record event types and provider names without storing raw prompts by default. Prompts may contain confidential business information, source code, personal data, or customer material.</p>
<p>Finally, evaluate models using representative tasks from your own organisation. Compare output quality, latency, token usage metadata when returned, and human review outcomes. An explicit evaluation set is more useful than assuming one provider is best for every workload. The adapter layer built here gives GCC teams a small, inspectable foundation for that comparison while retaining a clear path to more capable, controlled agent workflows.</p>
Top comments (0)