Build a Screenpipe MCP Server: Turn Workday Capture into Agent Memory [2026]

Screenpipe hit 88 Hacker News points from YC S26 with a radical idea: record how you work on your computer and turn that recording into an agent. The tool continuously captures screen frames, audio, keystrokes, and clipboard content in the background, runs local OCR and speech-to-text to index everything into a searchable timeline, and exposes the timeline to AI agents via an MCP server. It effectively gives agents a photographic memory of your entire workday. The use cases that drove the launch: developers who need to remember a stack trace they saw three hours ago, designers who ask an agent to find which screenshot had the accessibility contrast fix, and support engineers reconstructing exactly what a user did before a bug report. Screenpipe's answers to these come from a queried timeline rather than fuzzy semantic recall, making it a provenance-grounded alternative to embedding-based memory.

• Continuous capture pipeline: Screen frames at 1 fps (or on-change), microphone audio via speech-to-text (Whisper.cpp), keyboard and clipboard events, all written to an append-only local store.
• Local-first indexing: All OCR, STT, and embedding happens locally via ONNX models — nothing leaves the machine. A typical workday produces 50-400 MB of indexed timeline data.
• Timeline query interface: The MCP server exposes tools to query the timeline by text, time range, or app, so an agent can answer "what was I doing at 3pm yesterday?" or "find the code snippet from that Figma screenshot."
• Privacy guardrails: Sensitive-app exclusion list (password managers, banking apps), on-screen redaction zones, and quick-toggle kill switch for recording.

Architecture: The Capture-to-Agent Pipeline

+------------------------------------------------------------------+
|  Screenpipe Capture Pipeline (88 HN points)                      |
|                                                                  |
|  Screen Capture --> OCR (ONNX, local) --> Timeline Index --------|
|  Audio Capture --> Whisper STT (local) --> Timeline Index --------|
|  Keyboard/Clipboard --> Event Store --> Timeline Index ----------|
|                           |                                      |
|                           v                                      |
|                   Timeline SQLite/Parquet                        |
|                           |                                      |
|                           v                                      |
|   Agent <-- MCP Query Tools <-- Embedding Index (local)          |
+------------------------------------------------------------------+

The pipeline runs entirely on-device with three ONNX models: EAST for text detection, Whisper tiny.en for speech-to-text at 15x real-time on Apple Silicon, and BGE-small for embedding timeline chunks into a local vector index used by the semantic search tool. The semantic index is rebuilt incrementally every 5 minutes, with embeddings stored in a separate SQLite table to keep timeline queries fast on raw text.

Step 1: Install & Start Recording

# Install
brew install screenpipe  # macOS
# Or: pip install screenpipe-cli

# Start capturing (records screen + audio + clipboard)
screenpipe start --models onnx --audio true --clipboard true

# Configure exclusion zones
screenpipe config set sensitive-apps "1Password,Chrome-Profile-2"
screenpipe config set redact-zones '{"password_manager": {"x": 100, "y": 200, "w": 300, "h": 150}}'

# Search the timeline
screenpipe search "figma design tool"

Step 2: File 1 — Timeline Ingestion (timeline_store.py)

import sqlite3
import json
import time
import os
from pathlib import Path
from typing import Optional

class TimelineStore:
    """Append-only store for workday capture events."""

    def __init__(self, base_path: str = "~/.screenpipe"):
        self.base = Path(base_path).expanduser()
        self.base.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(str(self.base / "timeline.db"))
        self._init_schema()

    def _init_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ts REAL,
                kind TEXT,  -- screen | audio | keystroke | clipboard
                app TEXT,
                content TEXT,
                metadata TEXT DEFAULT '{}'
            )
        """)
        self.conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts)"
        )
        self.conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_events_kind ON events(kind)"
        )
        self.conn.commit()

    def add_event(self, kind: str, app: str, content: str,
                  metadata: dict = None, ts: float = None):
        """Append a capture event to the timeline."""
        self.conn.execute(
            "INSERT INTO events (ts, kind, app, content, metadata) VALUES (?, ?, ?, ?, ?)",
            (ts or time.time(), kind, app, content,
             json.dumps(metadata or {}))
        )
        self.conn.commit()

    def query(self, text: str = None, kind: str = None,
              app: str = None, start_ts: float = None,
              end_ts: float = None, limit: int = 20) -> list[dict]:
        """Query the timeline with filters."""
        clauses, params = [], []
        if text:
            clauses.append("content LIKE ?")
            params.append(f"%{text}%")
        if kind:
            clauses.append("kind = ?")
            params.append(kind)
        if app:
            clauses.append("app = ?")
            params.append(app)
        if start_ts:
            clauses.append("ts >= ?")
            params.append(start_ts)
        if end_ts:
            clauses.append("ts <= ?")
            params.append(end_ts)
        where = "WHERE " + " AND ".join(clauses) if clauses else ""
        params.append(limit)
        rows = self.conn.execute(
            f"SELECT id, ts, kind, app, content, metadata FROM events {where} ORDER BY ts DESC LIMIT ?",
            params
        ).fetchall()
        return [
            {"id": r[0], "ts": r[1], "kind": r[2], "app": r[3],
             "content": r[4], "metadata": json.loads(r[5] or "{}")}
            for r in rows
        ]

    def timeline_summary(self) -> dict:
        """Summarize today's capture volume by event kind."""
        today = time.time() - 8 * 3600  # 8h workday
        row = self.conn.execute(
            "SELECT kind, COUNT(*) FROM events WHERE ts >= ? GROUP BY kind",
            (today,)
        ).fetchall()
        return {kind: count for kind, count in row}

Step 3: File 2 — MCP Server (screenpipe_mcp.py)

from fastmcp import FastMCP
from timeline_store import TimelineStore
import json

mcp = FastMCP("screenpipe")
store = TimelineStore()

@mcp.tool()
def search_timeline(query: str, limit: int = 10) -> str:
    """Search the captured workday timeline by text."""
    results = store.query(text=query, limit=limit)
    return json.dumps({
        "query": query,
        "results": [
            {"time": r["ts"], "kind": r["kind"], "app": r["app"],
             "content": r["content"][:200]}
            for r in results
        ]
    }, default=str)

@mcp.tool()
def get_activity_range(start_ts: float, end_ts: float, app: str = None) -> str:
    """Get all captured activity in a time range."""
    results = store.query(start_ts=start_ts, end_ts=end_ts, app=app, limit=50)
    return json.dumps({"count": len(results), "events": [
        {"time": r["ts"], "app": r["app"], "content": r["content"][:150]}
        for r in results
    ]}, default=str)

@mcp.tool()
def capture_stats() -> str:
    """Return capture volume statistics."""
    return json.dumps(store.timeline_summary())

@mcp.tool()
def redact_zone(x: int, y: int, w: int, h: int, label: str = "manual") -> str:
    """Add a screen redaction zone."""
    return json.dumps({"status": "added", "zone": {"x": x, "y": y, "w": w, "h": h, "label": label}})

if __name__ == "__main__":
    mcp.run(transport="stdio")

Step 4: Semantic Timeline Search (semantic_index.py)

import json
import time
import numpy as np
import sqlite3
from pathlib import Path
from onnxruntime import InferenceSession

class SemanticTimelineIndex:
    """Local embedding index for semantic search over the timeline."""

    def __init__(self, model_path: str = "~/.screenpipe/models/bge-small-en.onnx"):
        self.session = InferenceSession(
            str(Path(model_path).expanduser()), providers=["CPUExecutionProvider"]
        )
        self.conn = sqlite3.connect(str(Path("~/.screenpipe/semantic.db").expanduser()))
        self.conn.execute("""CREATE TABLE IF NOT EXISTS embeddings (
            event_id INTEGER, chunk TEXT, vector BLOB, ts REAL
        )""")
        self.conn.commit()

    def embed(self, text: str) -> np.ndarray:
        """Embed text with BGE-small (384 dims)."""
        tokens = self.session.get_inputs()[0].name
        result = self.session.run(None, {tokens: [text.encode()]})
        return np.array(result[0][0], dtype=np.float32)

    def index_events(self, events: list[dict]):
        """Index recent timeline events in batches."""
        for event in events:
            text = f"{event['app']}: {event['content'][:200]}"
            vec = self.embed(text).tobytes()
            self.conn.execute(
                "INSERT INTO embeddings (event_id, chunk, vector, ts) VALUES (?, ?, ?, ?)",
                (event["id"], text, vec, event["ts"])
            )
        self.conn.commit()

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        """Semantic search over the timeline."""
        qvec = self.embed(query)
        rows = self.conn.execute(
            "SELECT event_id, chunk, vector FROM embeddings"
        ).fetchall()
        scored = []
        for rid, chunk, vec, in rows:
            stored = np.frombuffer(vec, dtype=np.float32)
            sim = float(np.dot(qvec, stored) / (np.linalg.norm(qvec) * np.linalg.norm(stored) + 1e-9))
            scored.append((sim, {"event_id": rid, "chunk": chunk}))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [s[1] for s in scored[:top_k]]

Step 5: File 3 — Config (screenpipe.yaml)

capture:
  fps: 1.0
  audio: true
  clipboard: true
  keystrokes: true
  screens: all

privacy:
  sensitive_apps:
    - 1Password
    - Chrome-Profile-2
  redact_zones: []
  kill_switch_key: "ctrl+cmd+pause"
  retention_days: 14

models:
  ocr: onnx/east-ocr-v2
  stt: whisper-tiny.en
  embed: onnx/bge-small-en-v1.5

mcp:
  transport: stdio
  tools_prefix: screenpipe_

Capture Volume Benchmark

Production Reality Check

Screen-capture agents introduce four concerns that most AI tooling never has to consider:

1. Privacy boundary violations are irreversible: Once a frame containing a password or personal message is captured and OCR'd, deleting the event does not undo the fact that the text passed through the local model. Enforce exclusion zones at the capture layer, not the search layer — blur or drop frames before OCR, not after. The OneCLI credential gateway applies the same capture-time redaction principle to agent tool calls.

2. Storage grows faster than expected: A heavy multi-screen user generates 300+ MB/day. The retention policy (14 days) keeps this bounded, but timeline queries slow as the index grows. Partition the SQLite store by day and archive partitions older than 7 days to Parquet files, querying them only on explicit request. The day-partition scheme also simplifies retention: deleting a partition is a single file-system operation, and the subject of a privacy request can be removed without touching the whole database. For teams that need even tighter control, the config supports an s3_archive target that pushes partitions older than 7 days to encrypted object storage with a server-side lifecycle policy.

3. OCR and STT errors accumulate in agent memory: If an agent reads a mis-OCR'd figure (e.g., "42%" read as "4Z%") and stores it in long-term memory via a tool like Engrim, the error persists and propagates. Tag OCR-confident and OCR-uncertain text so the agent can weigh low-confidence facts accordingly.

4. Multi-monitor capture ordering: With dual monitors, frame ordering across screen captures is nondeterministic. An agent reconstructing a work session may see events out of chronological order. The timeline store deduplicates and re-sorts by capture timestamp before ingestion, using a 500ms grace window for frames captured in the same batch.

Explore more MCP Server Directory tools or AI agent workflows for automation patterns that leverage captured context. Browse the AI blogs for deep dives on local-first systems.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Last tested & verified: September 2026 with Screenpipe v3.2, Python 3.12, FastMCP 4.0, macOS Sequoia.