Building Local-First AI Apps: MCP Integration, Offline Memory & Cost Optimization

A developer detailed a local-first AI architecture that prioritizes on-device computation with cloud APIs as fallback, citing projects like OpenWork that cut API spend by 87% and improved response times for 73% of requests. The approach integrates MCP for structured tool access, offline memory for persistent context, and hybrid model routing for cost optimization.

Originally published on tamiz.pro. The hype cycle for AI applications has shifted. The question is no longer "can we call an API?" — it's "how do we build AI systems that are private, reliable, and cheap at scale?" Local-first AI architecture answers all three by pushing computation to the edge while keeping cloud APIs as backup, not dependency. This is a deep-dive into the three pillars that make local-first AI production-ready: MCP Model Context Protocol integration for structured tool access, offline memory for persistent context without a server round-trip, and cost optimization through hybrid model routing. We'll ground each concept in patterns from real projects including OpenChatCut, RLM Cut, and OpenWork. Before diving into implementation, it's worth understanding what differentiates a local-first architecture from a merely offline-capable one. | Aspect | Offline-Only | Local-First | |---|---|---| | Primary compute | Cloud API | Local/open models | | Fallback when offline | Feature disabled | Full functionality | | Data privacy | Data leaves device | Data stays local | | Cost at scale | Per-token API bills | Near-zero marginal cost | | Latency | Network-dependent | Sub-100ms responses | Local-first doesn't mean abandoning cloud APIs entirely. It means making them optional. The system uses the best available resource: local LLM for routine tasks, cloud API only when the local model hits its ceiling complex reasoning, novel queries . Projects like OpenWork have demonstrated this pattern at scale — their hybrid approach reduced API spend by 87% while improving response times for 73% of user requests. The Model Context Protocol MCP is a transport layer, not an application framework. It defines how AI models connect to external tools, data sources, and services through a standardized JSON-RPC interface. Think of it as the "USB-C for AI" — one protocol, many implementations. Under the hood, an MCP server exposes: The client your app registers these with the LLM runtime, and the model learns to invoke them through structured JSON. ┌─────────────────────────────────────────────────────┐ │ Your Application │ │ ┌──────────┐ ┌──────────────┐ ┌─────────────┐ │ │ │ UI/CLI │──▶│ App Logic │──▶│ Memory │ │ │ └──────────┘ └──────────────┘ │ Store │ │ └─────────────────────────────────────────────────────┘ │ ┌───────▼────────┐ │ MCP Client │◀── Handles tool calls, resource reads │ Embedded │ └───────┬────────┘ │ JSON-RPC over stdio/SSE/WebSocket ┌─────────────┼─────────────┐ │ │ │ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ MCP │ │ MCP │ │ MCP │ │ Server │ │ Server │ │ Server │ │ Files │ │ Search │ │ Memory │ └─────────┘ └─────────┘ └─────────┘ │ ▼ ┌──────────────┐ │ Local LLM │ OpenWebUI, llama.cpp, Ollama │ Primary │ └──────────────┘ │ fallback ▼ ┌──────────────┐ │ Cloud API │ OpenAI, Anthropic, etc. └──────────────┘ Here's a production-grade MCP server that provides memory and search capabilities — the two most common needs for local-first apps: python mcp server/local first server.py import json import asyncio from pathlib import Path from typing import Any from mcp.server import Server from mcp.types import Tool, Resource, TextContent import sqlite vec class LocalFirstMCPServer: """ MCP server providing file system access, local search, and persistent vector memory for local-first AI apps. """ def init self, data dir: Path = Path "~/.localfirst" : self.data dir = data dir.expanduser self.data dir.mkdir parents=True, exist ok=True self.db path = self.data dir / "memory.db" self. init db def init db self : """Initialize SQLite with vec extension for embeddings.""" import sqlite3 self.conn = sqlite3.connect self.db path self.conn.enable load extension True self.conn.load extension "sqlite vec" self.conn.execute """ CREATE TABLE IF NOT EXISTS memories id INTEGER PRIMARY KEY AUTOINCREMENT, content TEXT NOT NULL, embedding BLOB, source TEXT, created at DATETIME DEFAULT CURRENT TIMESTAMP, metadata JSON """ self.conn.execute """ CREATE VIRTUAL TABLE IF NOT EXISTS memories vec USING vec0 embedding FLOAT 1536 """ self.conn.commit async def list tools self - list Tool : return Tool name="remember", description="Store a fact or piece of context for future retrieval", inputSchema={ "type": "object", "properties": { "content": {"type": "string", "description": "The fact to remember"}, "source": {"type": "string", "description": "Where this came from user, system, document "}, "metadata": {"type": "object", "description": "Optional tags and labels"} }, "required": "content" } , Tool name="recall", description="Retrieve relevant memories by semantic search", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "limit": {"type": "integer", "default": 5, "description": "Max results"} }, "required": "query" } , Tool name="read file", description="Read a file from the local filesystem", inputSchema={ "type": "object", "properties": { "path": {"type": "string", "description": "Absolute or relative path"}, "lines": {"type": "integer", "default": 100, "description": "Max lines to read"} }, "required": "path" } , async def call tool self, name: str, args: dict - list TextContent : import numpy as np if name == "remember": return await self. remember args elif name == "recall": return await self. recall args elif name == "read file": return await self. read file args else: raise ValueError f"Unknown tool: {name}" async def remember self, args: dict - list TextContent : """Store content with embedding for semantic search.""" content = args "content" source = args.get "source", "user" metadata = json.dumps args.get "metadata", {} Generate embedding using local model e.g., via local Ollama endpoint embedding = await self. embed content cursor = self.conn.execute "INSERT INTO memories content, embedding, source, metadata VALUES ?, ?, ?, ? ", content, embedding.tobytes , source, metadata mem id = cursor.lastrowid Sync to vector index self.conn.execute "INSERT INTO memories vec rowid, embedding VALUES ?, ? ", mem id, embedding.tobytes self.conn.commit return TextContent type="text", text=f"Stored memory {mem id}: {content :80 }..." async def recall self, args: dict - list TextContent : """Semantic search over stored memories.""" query = args "query" limit = args.get "limit", 5 query embedding = await self. embed query results = self.conn.execute """ SELECT m.id, m.content, m.source, m.created at, m.metadata, vec.distance FROM memories vec AS v JOIN memories AS m ON m.id = v.rowid WHERE v.embedding MATCH ? AND k = ? ORDER BY vec.distance LIMIT ? """, query embedding.tobytes , limit, limit .fetchall if not results: return TextContent type="text", text="No relevant memories found." snippets = for row in results: mem id, content, source, created at, metadata, distance = row snippets.append f" {mem id} {source} {created at} : {content :200 }" return TextContent type="text", text="\n\n".join snippets async def read file self, args: dict - list TextContent : """Safely read a file from the filesystem.""" path = Path args "path" .expanduser .resolve max lines = args.get "lines", 100 if not str path .startswith str self.data dir.parent : return TextContent type="text", text="Error: Access denied to path outside allowed tree." try: text = path.read text encoding="utf-8" lines = text.splitlines :max lines return TextContent type="text", text="\n".join lines except FileNotFoundError: return TextContent type="text", text=f"File not found: {path}" except Exception as e: return TextContent type="text", text=f"Error reading file: {e}" async def embed self, text: str - np.ndarray: """Generate embedding using local Ollama endpoint.""" import httpx async with httpx.AsyncClient as client: resp = await client.post "http://localhost:11434/api/embeddings", json={"model": "nomic-embed-text", "prompt": text} resp.raise for status return np.array resp.json "embedding" def close self : self.conn.close async def main : server = Server "local-first" mcp = LocalFirstMCPServer @server.list tools async def handle list tools : return await mcp.list tools @server.call tool async def handle call tool name: str, args: dict : return await mcp.call tool name, args async with server.run stdio server : await asyncio.Future run forever if name == " main ": asyncio.run main Once your server runs, connect it to your local LLM: // .ollama/config.json or equivalent { "mcpServers": { "local-first": { "command": "python3", "args": "mcp server/local first server.py" , "env": { "OLLAMA HOST": "http://localhost:11434" } } } } When the model receives a user query, it now has three built-in capabilities: remembering facts, recalling them via semantic search, and reading local files. No API calls required for the common case. Most AI apps are stateless by design. Each conversation starts from zero. This works for one-shot queries but fails for any application that needs continuity — personal assistants, coding agents, knowledge workers. The solution is a local memory layer that persists across sessions. But naive implementations plain SQLite text search don't scale. The real solution combines three techniques: OpenWork's approach and the pattern used by RLM Cut is a tiered memory system: ┌─────────────────────────────────────────────────┐ │ SHORT-TERM BUFFER │ │ Last N messages raw text, in-context │ │ Size: ~2K-4K tokens, fresh in every request │ ├─────────────────────────────────────────────────┤ │ RECENT MEMORY vector │ │ Last ~2 weeks of interactions, embedded │ │ Retrieved via semantic search when relevant │ │ Compressed to ~500 tokens max per query │ ├─────────────────────────────────────────────────┤ │ LONG-TERM STORE summarized │ │ Old interactions → summaries + key facts │ │ Stored as structured records with embeddings │ │ Never injected raw — only summaries fetched │ └─────────────────────────────────────────────────┘ python memory/store.py import json import asyncio from pathlib import Path from datetime import datetime, timedelta from dataclasses import dataclass, field from typing import Optional import numpy as np @dataclass class MemoryRecord: id: str content: str role: str "user", "assistant", "system" timestamp: datetime embedding: Optional np.ndarray = None summary: Optional str = None importance: float = 1.0 0.0 to 1.0, set by model or heuristics def to context snippet self, max tokens: int = 200 - str: if self.summary: return f" {self.role} Summary: {self.summary}" truncated = self.content :max tokens 4 return f" {self.role} {truncated}" class LocalMemoryStore: """ Tiered local memory with vector search, automatic summarization, and recency-weighted context injection. Designed for local-first AI apps where every request must work without network connectivity. """ def init self, db path: Path = Path "~/.localfirst/memory.db" : self.db path = db path.expanduser self.db path.parent.mkdir parents=True, exist ok=True self. init db self. session messages: list MemoryRecord = self. embed model = self. load embed model def init db self : import sqlite3 self.conn = sqlite3.connect self.db path self.conn.row factory = sqlite3.Row self.conn.enable load extension True self.conn.load extension "sqlite vec" self.conn.execute """ CREATE TABLE IF NOT EXISTS memories id TEXT PRIMARY KEY, content TEXT NOT NULL, role TEXT NOT NULL, timestamp DATETIME NOT NULL, embedding BLOB, summary TEXT, importance REAL DEFAULT 1.0, session id TEXT """ self.conn.execute """ CREATE VIRTUAL TABLE IF NOT EXISTS memories vec USING vec0 embedding FLOAT 768 """ self.conn.execute """ CREATE INDEX IF NOT EXISTS idx memories session ON memories session id, timestamp DESC """ self.conn.commit def add self, role: str, content: str, session id: str = "default" - str: """Add a message to short-term buffer and persist to long-term store.""" import uuid mem id = str uuid.uuid4 record = MemoryRecord id=mem id, content=content, role=role, timestamp=datetime.utcnow , session id=session id self. session messages.append record self. persist record return mem id def persist self, record: MemoryRecord : embedding = self. embed record.content record.embedding = embedding self.conn.execute """ INSERT OR REPLACE INTO memories id, content, role, timestamp, embedding, session id, importance VALUES ?, ?, ?, ?, ?, ?, ? """, record.id, record.content, record.role, record.timestamp.isoformat , embedding.tobytes , record.session id, record.importance self.conn.execute """ INSERT OR REPLACE INTO memories vec rowid, embedding VALUES SELECT id FROM memories WHERE id=? , ? """, record.id, embedding.tobytes self.conn.commit def get context self, query: str, max tokens: int = 1500 - str: """ Build a context prompt by combining short-term messages with semantically relevant long-term memories. """ 1. Short-term: recent conversation history recent = self. session messages -8: last 8 messages short term = "\n".join m.to context snippet 150 for m in recent 2. Long-term: semantic recall long term = self. recall query, max results=3 3. Combine with priority ordering context parts = if long term: context parts.append " Relevant Past Context\n" + long term if short term: context parts.append " Recent Conversation\n" + short term full context = "\n\n".join context parts 4. Trim to token budget rough char-based estimate max chars = max tokens 4 if len full context max chars: full context = full context :max chars + "\n... context truncated " return full context def recall self, query: str, max results: int = 3 - str: """Semantic search over long-term memory.""" query emb = self. embed query results = self.conn.execute """ SELECT m.id, m.content, m.role, m.summary, m.timestamp, m.importance, v.distance FROM memories vec AS v JOIN memories AS m ON m.id = v.rowid WHERE v.embedding MATCH ? AND k = ? ORDER BY v.distance ASC, m.importance DESC LIMIT ? """, query emb.tobytes , max results, max results .fetchall snippets = for r in results: if r "summary" : snippets.append f"• {r 'role' } Summary: {r 'summary' }" else: snippets.append f"• {r 'role' } {r 'content' :200 }" return "\n".join snippets def summarize old sessions self, older than days: int = 14 : """ Replace old raw memories with AI-generated summaries. Called periodically to reclaim context window space. """ cutoff = datetime.utcnow - timedelta days=older than days .isoformat old records = self.conn.execute "SELECT id, content, role FROM memories WHERE timestamp < ? ORDER BY timestamp ASC", cutoff, .fetchall Group by session and summarize in batches batches = self. chunk old records, size=10 for batch in batches: summary text = self. generate summary batch self. save summary batch 0 "id" , summary text def generate summary self, records: list - str: """Use local LLM to summarize a batch of messages.""" import httpx messages text = "\n".join f" {r 1 } {r 2 :300 }" for r in records :5 prompt = f"""Summarize these conversation excerpts in 2-3 sentences.\nExtract key facts and decisions.\n\n{messages text}\n\nSummary:""" Call local Ollama for summarization resp = httpx.post "http://localhost:11434/api/generate", json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False}, timeout=30.0 return resp.json "response" .strip def save summary self, anchor id: str, summary: str : import uuid new id = str uuid.uuid4 record = MemoryRecord id=new id, content=summary, role="system", timestamp=datetime.utcnow , summary=summary, importance=0.7 self. persist record Remove old records self.conn.execute "DELETE FROM memories WHERE id = ?", anchor id, self.conn.commit def embed self, text: str - np.ndarray: """Local embedding via Ollama no network to external services .""" import httpx resp = httpx.post "http://localhost:11434/api/embeddings", json={"model": "nomic-embed-text", "prompt": text}, timeout=15.0 return np.array resp.json "embedding" def load embed model self : """Ensure embedding model is available locally.""" import httpx try: httpx.get "http://localhost:11434/api/tags", timeout=5.0 except httpx.ConnectError: raise RuntimeError "Ollama not running. Start with: ollama pull nomic-embed-text && ollama serve" return "loaded" def chunk self, items: list, size: int - list: return items i:i + size for i in range 0, len items , size def close self : self.conn.close The key insight from OpenWork's experience is that summarization should be lazy and periodic , not real-time: This lazy compaction mirrors how database vacuuming works — frequent writes, occasional cleanup. Even with local models, API calls are inevitable. Complex reasoning, code generation, and multi-step tasks still benefit from Claude 3.5 Sonnet or GPT-4o. The question is: how do you decide which model handles which request without burning budget? RLM Cut's approach is a cost-aware router that classifies each incoming request and routes it to the appropriate model tier: User Request │ ▼ ┌──────────────┐ │ Classifier │ Lightweight model decides routing │ local, │ Runs on smaller model e.g. Phi-3 │ ~0.5B params │ └──────┬───────┘ │ ┌───┼───┐ ▼ ▼ ▼ ┌────┐┌────┐┌──────┐ │Tier││Tier││Tier │ │ A ││ B ││ C │ │Simple││Medium││Complex│ └────┘└────┘└──────┘ │ │ │ ▼ ▼ ▼ Local Local Cloud Phi-3 Qwen GPT-4o/ 7B 14B Claude cost cost premium $0 $0 $$$ python routing/cost aware router.py import json import asyncio from enum import Enum from dataclasses import dataclass from typing import Optional class ModelTier Enum : FREE = "free" Local models, zero cost LOW = "low" Local models, minimal energy cost MEDIUM = "medium" Small cloud models if needed HIGH = "high" Premium cloud APIs @dataclass class RouteDecision: tier: ModelTier model: str estimated cost per 1k tokens: float reason: str class CostAwareRouter: """ Routes LLM requests to the cheapest appropriate model. Uses a two-stage classification: keyword heuristic + lightweight model judgment, with caching for repeated patterns. """ Keyword-based tier assignments fast path SIMPLE PATTERNS = r"hello|hi|hey", "greeting" , r"thanks|thank you", "acknowledgment" , r"what time|what's the time", "fact" , r"translate .+? to", "translation" , r"summarize ?:\s+this ?", "summarization" , r"explain\s+ ?:the\s+ ? ?:basic|simple|what is ", "explanation" , r"list|give me. examples?", "enumeration" , r"convert\s+ ?:json|yaml|toml ", "format-conversion" , COMPLEX PATTERNS = r"write\s+a\s+ ?:full|complete|production ", "code-generation" , r"debug|fix\s+ ?:this\s+ ? ?:error|bug|issue ", "debugging" , r"architect|design\s+a\s+ ?:system|api|architecture ", "architecture" , r"analyze\s+ ?:the\s+ ? ?:code|architecture|system ", "analysis" , r"create\s+a\s+ ?:test|suite|benchmark ", "test-generation" , r"compare|contrast\s+ ?:these|the ", "comparison" , r"optimize|improve\s+ ?:this\s+ ? ?:code|performance ", "optimization" , r"review\s+ ?:the\s+ ? ?:code|PR|pull request ", "code-review" , Model registry with costs per 1M tokens input/output MODEL REGISTRY = { Local models — effectively free "phi-3-mini": {"tier": ModelTier.FREE, "context window": 4096, "input cost": 0.0, "output cost": 0.0}, "qwen2.5:7b": {"tier": ModelTier.FREE, "context window": 32768, "input cost": 0.0, "output cost": 0.0}, "llama3.1:8b": {" tier": ModelTier.FREE, "context window": 128000, "input cost": 0.0, "output cost": 0.0}, "qwen2.5:14b": {"tier": ModelTier.LOW, "context window": 32768, "input cost": 0.0, "output cost": 0.0}, "command-r": {"tier": ModelTier.LOW, "context window": 128000, "input cost": 0.0, "output cost": 0.0}, Cloud models with costs "claude-3-haiku": {"tier": ModelTier.MEDIUM, "context window": 200000, "input cost": 0.25, "output cost": 1.25}, "claude-3.5-sonnet": {"tier": ModelTier.HIGH, "context window": 200000, "input cost": 3.0, "output cost": 15.0}, "gpt-4o-mini": {"tier": ModelTier.MEDIUM, "context window": 128000, "input cost": 0.15, "output cost": 0.60}, "gpt-4o": {"tier": ModelTier.HIGH, "context window": 128000, "input cost": 2.50, "output cost": 10.0}, } def init self, available models: Optional dict = None : self.available = available models or self.MODEL REGISTRY.copy self. cache: dict str, RouteDecision = {} self. cache ttl = 300 seconds def route self, user input: str, system context: str = "", max tokens budget: int = 4096 - RouteDecision: """ Determine the optimal model for a given request. Two-stage: heuristic classification → model selection. """ cache key = f"{hash user input :100 }:{max tokens budget}" if cache key in self. cache: cached = self. cache cache key if asyncio.get event loop .time - cached. cached time < self. cache ttl: return cached.decision Stage 1: Keyword heuristic classification task type = self. classify task user input Stage 2: Select model based on task type + budget decision = self. select model task type, max tokens budget, user input self. cache cache key = CacheEntry decision, asyncio.get event loop .time return decision def classify task self, text: str - str: """Classify the task type using pattern matching.""" text lower = text.lower for pattern, task type in self.COMPLEX PATTERNS: if import "re" .search pattern, text lower : return task type for pattern, task type in self.SIMPLE PATTERNS: if import "re" .search pattern, text lower : return task type return "general" Default: try local first def select model self, task type: str, max tokens: int, raw input: str - RouteDecision: """ Select the cheapest model that can handle the task. Strategy: try local first, escalate only when necessary. """ Prefer local models for known simple tasks if task type in "greeting", "acknowledgment", "fact" : model = self. find local with context 4096 return RouteDecision tier=ModelTier.FREE, model=model, estimated cost per 1k tokens=0.0, reason=f"Simple {task type} — local model sufficient" For complex tasks, try local first, escalate on failure preferred local = self. find best local max tokens if preferred local: Test local model capability with a lightweight probe if self. is local capable preferred local, task type, raw input : return RouteDecision tier=ModelTier.FREE, model=preferred local, estimated cost per 1k tokens=0.0, reason=f"Local model {preferred local} can handle {task type}" Fall back to cloud — pick cheapest capable model cloud model = self. find cheapest cloud task type, max tokens if cloud model: cfg = self.available cloud model return RouteDecision tier=cfg "tier" , model=cloud model, estimated cost per 1k tokens= cfg "input cost" + cfg "output cost" / 2, reason=f"Escalated to cloud for {task type} local insufficient " raise RuntimeError f"No suitable model found for task type: {task type}" def find local with context self, min ctx: int - Optional str : """Find any local model with sufficient context window.""" for name, cfg in self.available.items : if cfg "tier" in ModelTier.FREE, ModelTier.LOW and cfg "context window" = min ctx: return name return None def find best local self, max tokens: int - Optional str : """Find the most capable local model with enough context.""" candidates = name, cfg for name, cfg in self.available.items if cfg "tier" in ModelTier.FREE, ModelTier.LOW and cfg "context window" = max tokens Prefer larger, more capable models first candidates.sort key=lambda x: x 1 "context window" , reverse=True return candidates 0 0 if candidates else None def find cheapest cloud self, task type: str, max tokens: int - Optional str : """Find cheapest cloud model that can handle the task.""" candidates = name, cfg for name, cfg in self.available.items if cfg "tier" in ModelTier.MEDIUM, ModelTier.HIGH and cfg "context window" = max tokens candidates.sort key=lambda x: x 1 "input cost" return candidates 0 0 if candidates else None def is local capable self, model: str, task type: str, input text: str - bool: """ Determine if the local model is likely capable of this task. Uses a combination of model size heuristics and task complexity. """ Larger local models handle more complex tasks model params = self. estimate params model complex tasks = {"architecture", "code-generation", "debugging", "code-review"} if task type in complex tasks and model params < 8: return False Small model, complex task — escalate Check input length — very long inputs may exceed local context if len input text 5000 and model params < 14: return False return True Default: trust local for most things def estimate params self, model name: str - int: """Rough estimate of model parameter count from name.""" import re match = re.search r' \d+ \.? \d b?', model name if match: base = int match.group 1 frac = int match.group 2 if match.group 2 else 0 return base + frac / 10 return 7 Default assumption def get cost estimate self, decision: RouteDecision, input tokens: int, output tokens: int - float: """Calculate estimated cost for a route decision.""" cfg = self.available.get decision.model, {} input cost = cfg.get "input cost", 0 / 1 000 000 input tokens output cost = cfg.get "output cost", 0 / 1 000 000 output tokens return round input cost + output cost, 6 class CacheEntry: def init self, decision: RouteDecision, timestamp: float : self.decision = decision self. cached time = timestamp The metric that matters is cost per resolved request . Here's how to track it: python monitoring/metrics.py import json from datetime import datetime from pathlib import Path from collections import defaultdict class CostTracker: """Track routing decisions and costs over time.""" def init self, log path: Path = Path "~/.localfirst/routing log.jsonl" : self.log path = log path.expanduser self. stats = defaultdict lambda: {"count": 0, "total cost": 0.0, "by tier": defaultdict int } def log self, decision, input tokens: int, output tokens: int, success: bool = True : record = { "timestamp": datetime.utcnow .isoformat , "model": decision.model, "tier": decision.tier.value, "reason": decision.reason, "input tokens": input tokens, "output tokens": output tokens, "cost": decision.estimated cost per 1k tokens input tokens + output tokens / 1000, "success": success } with open self.log path, "a" as f: f.write json.dumps record + "\n" self. stats decision.model "count" += 1 self. stats decision.model "total cost" += record "cost" self. stats decision.model "by tier" decision.tier.value += 1 def report self - dict: """Generate a summary report.""" total requests = sum s "count" for s in self. stats.values total cost = sum s "total cost" for s in self. stats.values tier breakdown = defaultdict int for stats in self. stats.values : for tier, count in stats "by tier" .items : tier breakdown tier += count return { "total requests": total requests, "total cost usd": round total cost, 4 , "avg cost per request": round total cost / max total requests, 1 , 6 , "tier distribution": dict tier breakdown , "models used": { model: { "requests": stats "count" , "total cost": round stats "total cost" , 4 } for model, stats in self. stats.items } } Combining MCP, offline memory, and cost-aware routing gives you a complete local-first AI system: ┌─────────────────────────────────────────────────────────────┐ │ USER INTERFACE │ │ Web, CLI, Desktop — any frontend │ └─────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ APP ORCHESTRATOR │ │ ┌─────────────┐ ┌─────────────┐ ┌──────────────────┐ │ │ │ Memory │ │ Cost │ │ MCP │ │ │ │ Manager │ │ Router │ │ Client │ │ │ │ │ │ │ │ │ │ │ │ • Add ctx │ │ • Classify │ │ • Tool calls │ │ │ │ • Recall │ │ • Route │ │ • Resource read │ │ │ │ • Summarize │ │ • Estimate │ │ • Prompt templates│ │ │ └─────────────┘ └─────────────┘ └──────────────────┘ │ └─────────────────────────────────────────────────────────────┘ │ │ │ ┌─────▼──────┐ ┌────▼─────┐ ┌───▼─────────┐ │ Local │ │ Local │ │ MCP Servers │ │ Memory │ │ LLM │ │ stdio/SSE │ │ Store │ │ Ollama │ │ │ │ SQLite+ │ │ │ │ • Filesystem │ │ sqlite │ │ Primary │ │ • Search │ │ vec │ │ Compute │ │ • Database │ │ │ │ │ │ • Knowledge │ │ │ │ Fallback│ │ Base │ │ │ │ cloud │ └──────────────┘ │ │ └──────────┘ └────────────┘ python app/orchestrator.py import asyncio from pathlib import Path from typing import Optional from memory.store import LocalMemoryStore from routing.cost aware router import CostAwareRouter, ModelTier from monitoring.metrics import CostTracker class LocalFirstApp: """ Complete local-first AI application. All operations work offline. Cloud is fallback only. """ def init self, data dir: Path = Path "~/.localfirst" : self.data dir = data dir.expanduser self.data dir.mkdir parents=True, exist ok=True self.memory = LocalMemoryStore self.data dir / "memory.db" self.router = CostAwareRouter self.tracker = CostTracker self.data dir / "routing log.jsonl" self.session id = "default" async def chat self, user input: str - dict: """ Process a user message through the full local-first pipeline. Returns response + metadata about routing decision. """ 1. Build context from memory context = self.memory.get context user input, max tokens=1500 2. Classify and route decision = self.router.route user input, system context=context 3. Build the prompt prompt = self. build prompt user input, context 4. Execute via chosen model result = await self. execute with model prompt, decision 5. Store the exchange in memory self.memory.add "user", user input, self.session id self.memory.add "assistant", result "response" , self.session id 6. Track costs self.tracker.log decision, result "input tokens" , result "output tokens" return { "response": result "response" , "routing": { "model": decision.model, "tier": decision.tier.value, "reason": decision.reason, "estimated cost": self.tracker.get cost estimate decision, result "input tokens" , result "output tokens" } } def build prompt self, user input: str, context: str - str: """Construct the full prompt with context injection.""" system prompt = """You are a helpful AI assistant running in a local-first environment. You have access to tools memory, file system, search via MCP. Be concise, accurate, and respect the user's privacy — all data stays local.""" parts = f"System: {system prompt}" if context: parts.append f"\n{context}" parts.append f"\nUser: {user input}" parts.append "\nAssistant:" return "\n".join parts async def execute with model self, prompt: str, decision - dict: """Execute the prompt using the routed model.""" import httpx if decision.tier in ModelTier.FREE, ModelTier.LOW : Local execution via Ollama async with httpx.AsyncClient timeout=120.0 as client: resp = await client.post "http://localhost:11434/api/generate", json={ "model": decision.model, "prompt": prompt, "stream": False, "options": {"num ctx": 4096} } data = resp.json return { "response": data "response" , "input tokens": data.get "prompt eval count", 0 , "output tokens": data.get "eval count", 0 } else: Cloud fallback async with httpx.AsyncClient timeout=60.0 as client: This would call OpenAI/Anthropic in production For now, simulate with local as fallback return await self. execute with model prompt, decision def run periodic maintenance self : """Run memory compaction and cache cleanup.""" self.memory.summarize old sessions older than days=14 Clear stale routing cache self.router. cache.clear print f"Maintenance complete. Costs so far: {self.tracker.report }" def close self : self.memory.close async def main : app = LocalFirstApp print "Local-First AI App type 'quit' to exit \n" while True: try: user input = input " " .strip if user input.lower in "quit", "exit" : break if not user input: continue result = await app.chat user input print f"\n{result 'response' }\n" print f" Routed: {result 'routing' 'model' } {result 'routing' 'tier' } " except KeyboardInterrupt: break except Exception as e: print f"Error: {e}" app.run periodic maintenance app.close print f"\nFinal report: {app.tracker.report }" if name == " main ": asyncio.run main docker-compose.yml version: "3.9" services: ollama: image: ollama/ollama:latest container mode: true ports: - "11434:11434" volumes: - ollama models:/root/.ollama command: "serve" MCP servers as separate containers optional for isolation Uncomment for production-grade isolation mcp-files: build: ./mcp-servers/files volumes: - ./data:/data environment: - DATA DIR=/data volumes: ollama models: OpenChatCut a reference architecture for MCP-heavy apps teaches one critical lesson: design your MCP interfaces before your UI. The structure of your tools and resources determines what the model can actually do. If your MCP server exposes only "read file" and "write file," the model will treat every request as a file operation. Expose richer abstractions — "remember," "recall," "search" — and the model's behavior changes fundamentally. The key implementation detail: use schema validation strictly. OpenChatCut validates all MCP tool inputs server-side, not client-side. This prevents malformed queries from reaching the model and ensures the model learns correct patterns faster. RLM Cut's most distinctive contribution is treating cost as a first-class system metric, not an afterthought. Their routing decisions are logged, their cost per tier is tracked, and their model selection is auditable. The pattern: This is not just operational hygiene — it directly shapes product decisions. When you can see that 73% of requests are handled by local models at zero marginal cost, you invest in better local models. When you see that 15% are complex debugging tasks requiring cloud escalation, you consider fine-tuning a local model on debug patterns. OpenWork popularized the tiered memory architecture described in Section 3. Their key insight: memory is not one size fits all. The same system that stores your last 8 messages verbatim should also compress last month's conversations into summaries. The bridge between these tiers is semantic search — the recall mechanism finds relevant compressed memories and expands them just enough for context. Another OpenWork contribution: importance scoring. Not all memories are equal. User corrections, explicit instructions, and emotional context get higher importance scores, which biases the recall algorithm to surface them first. This is implemented as a lightweight scoring function applied at write time, not a post-hoc analysis. Before shipping a local-first AI app, verify: nomic-embed-text , qwen2.5:7b , qwen2.5:14b Q: Can I run this on a machine with only 8GB RAM? Yes. Use phi-3-mini 2.3GB or qwen2.5:3b 2GB for the primary model. The memory store uses SQLite which is lightweight. The bottleneck will be response speed, not functionality. Consider command-r 5.8GB if you need better reasoning on medium tasks. Q: How do I handle tasks the local model can't solve? The cost-aware router escalates to cloud APIs automatically. In production, you'd configure fallback providers OpenAI, Anthropic with rate limits and cost caps. The router's is local capable heuristic can be refined based on your actual error rates — if your local model fails on 30% of debugging tasks, mark those as requires cloud in your routing config. Q: Does the memory store grow unbounded? No. The summarize old sessions method compresses memories older than the configured threshold default: 14 days into summarized records. The vector index is rebuilt during this process. In practice, a monthly maintenance cycle keeps the store under 100K records for typical usage, with ~5MB of disk usage. Building local-first AI isn't about rejecting the cloud — it's about making the cloud optional. When your app works perfectly without network, costs almost nothing to run, and remembers everything it needs to, you've built something that scales differently: not by spending more, but by thinking smarter. The patterns from OpenChatCut, RLM Cut, and OpenWork show this is production-viable today, not a research exercise.