email-insights

Un servidor MCP que expone análisis de señales de correo electrónico a Claude Desktop, con un trabajador en segundo plano para trabajos de extracción asíncronos programados y registro estructurado.

Estructura del proyecto

email-insights/
├── data/
│   └── emails.csv              # Raw email data (id, from, subject, body, date)
├── database/
│   └── signals.db              # SQLite database (created after running ingestion)
├── db/
│   ├── connection.py           # Single source of truth for SQLite connections
│   ├── schema.py               # DDL for all tables (idempotent CREATE IF NOT EXISTS)
│   ├── signals.py              # Read/write for signals table
│   ├── raw_emails.py           # Read/write for raw_emails table
│   └── jobs.py                 # Read/write for jobs and failed_extractions tables
├── ingestion/
│   ├── fetch_emails_imap.py    # Fetch emails via IMAP → store raw in SQLite
│   ├── parse_csv.py            # Step 1: Load emails from CSV
│   ├── extract_signals.py      # Step 2: Call local LLM to extract signals
│   └── store_signals.py        # Step 3: Write signals to SQLite (run this)
├── logs/
│   └── worker.log              # Rotating log file (auto-created, 5 MB max, 3 backups)
├── mcp_server/
│   ├── server.py               # MCP server: registers tools and starts listening
│   └── tools.py                # SQLite query functions + job scheduling tools
├── utils/
│   └── logger.py               # Shared structured logger (stderr + rotating file)
├── worker/
│   └── job_runner.py           # Background worker: polls SQLite and runs extraction jobs
├── requirements.txt
└── README.md

Related MCP server: io.github.p-w-4-z/inbox-mcp

Configuración

1. Instalar dependencias

pip install -r requirements.txt

2. Configurar las credenciales IMAP

Copia .env.example a .env y rellena tus credenciales:

IMAP_HOST=imap.gmail.com
IMAP_USER=you@gmail.com
IMAP_PASSWORD=your-app-specific-password
IMAP_PORT=993          # optional, default 993
IMAP_MAILBOX=INBOX     # optional, default INBOX

Para Gmail, genera una contraseña específica de la aplicación en myaccount.google.com/apppasswords.

3. Obtener correos electrónicos en SQLite

Obtén todos los correos electrónicos de tu bandeja de entrada y guárdalos en la tabla raw_emails:

python ingestion/fetch_emails_imap.py

Una barra de progreso muestra el estado de obtención y almacenamiento en tiempo real. Opciones:

# Fetch only the 50 most recent emails
python ingestion/fetch_emails_imap.py --limit 50

# Also export a CSV backup
python ingestion/fetch_emails_imap.py --output data/backup.csv

# Count emails in a date range (no fetch)
python ingestion/fetch_emails_imap.py --count --start-date 2025-01-01 --end-date 2025-03-01

4. Iniciar LM Studio

• Abre LM Studio y carga cualquier modelo que siga instrucciones (Llama 3, Mistral, etc.)

• Inicia el servidor local: Local Server → Start Server

• URL predeterminada: http://127.0.0.1:10101

• Copia la cadena del identificador del modelo y pégala en ingestion/extract_signals.py como LOCAL_MODEL

5. Ejecutar la extracción de señales

python ingestion/store_signals.py

Esto lee data/emails.csv, envía cada correo electrónico a tu LLM local para la extracción de señales

y almacena los resultados en database/signals.db.

6. Iniciar el trabajador en segundo plano

El trabajador es un proceso independiente que busca trabajos de extracción programados. Ejecútalo en una terminal dedicada:

python worker/job_runner.py

El trabajador registra toda la actividad en logs/worker.log y en stderr. Consulta SQLite cada 10 segundos y recoge automáticamente cualquier trabajo pendiente o programado.

7. Conectar Claude Desktop

Añade este servidor a tu configuración de Claude Desktop:

Mac: ~/Library/Application Support/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "email-insights": {
      "command": "python",
      "args": ["/absolute/path/to/email-insights/mcp_server/server.py"]
    }
  }
}

Reinicia Claude Desktop. Deberías ver email-insights en la lista de herramientas.

Herramientas MCP

Herramientas de consulta

Herramientas de programación de trabajos

Todas las herramientas de programación devuelven una respuesta inmediatamente. La extracción se ejecuta de forma asíncrona en el proceso del trabajador.

Modos de ejecución de schedule_extraction_tool

Arquitectura

Claude Desktop ──stdio──▶ mcp_server/server.py
                                  │
                          mcp_server/tools.py
                                  │
                           SQLite signals.db
                                  │
                        worker/job_runner.py  ◀── runs separately
                                  │
                          LM Studio (local LLM)

El servidor MCP y el trabajador son dos procesos completamente separados que solo comparten la base de datos SQLite. El servidor MCP nunca espera a que termine la extracción: crea un registro de trabajo y devuelve una respuesta inmediatamente. El trabajador posee todas las escrituras en las tablas jobs y failed_extractions (actualizaciones de estado, progreso, fallos); el servidor MCP solo lee el estado del trabajo.

Esquema de SQLite

CREATE TABLE raw_emails (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    email_id     TEXT    UNIQUE,    -- SHA-256(date|sender_name|sender_email)[:16]
    date         TEXT,              -- ISO format from email Date header
    sender_name  TEXT,
    sender_email TEXT,
    subject      TEXT,
    body         TEXT,
    fetched_at   TEXT DEFAULT (datetime('now'))
);

CREATE TABLE signals (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    email_id        TEXT    UNIQUE,
    topic           TEXT,       -- job application | recruiter outreach | rejection | interview | networking | other
    tone            TEXT,       -- positive | neutral | negative
    sender_type     TEXT,       -- recruiter | company HR | networking contact | university | other
    urgency         TEXT,       -- high | medium | low
    requires_action INTEGER,    -- 0 or 1
    date            TEXT        -- ISO format: YYYY-MM-DD
);

CREATE TABLE jobs (
    job_id           INTEGER PRIMARY KEY AUTOINCREMENT,
    schema_id        INTEGER,
    status           TEXT NOT NULL DEFAULT 'pending',  -- pending | scheduled | running | completed | failed
    run_at           TEXT,       -- ISO datetime; NULL means run immediately
    total_emails     INTEGER DEFAULT 0,
    processed_emails INTEGER DEFAULT 0,
    created_at       TEXT DEFAULT (datetime('now')),
    completed_at     TEXT,
    error_message    TEXT,
    retry_of_job_id  INTEGER     -- set for retry jobs; links back to source job
);

CREATE TABLE failed_extractions (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    job_id        INTEGER NOT NULL,
    email_id      TEXT NOT NULL,
    error_message TEXT,
    created_at    TEXT DEFAULT (datetime('now'))
);

Tanto jobs como failed_extractions se crean automáticamente en el primer uso; no se necesita migración manual.

Registro estructurado

Toda la actividad del trabajador se escribe en logs/worker.log (creado automáticamente) y en stderr.

Formato de registro:

[2026-03-05 14:22:01] [INFO] Worker started, polling every 10 seconds
[2026-03-05 14:22:11] [INFO] Job 1 picked up: schema_id=None, 10 emails to process
[2026-03-05 14:22:13] [INFO] [1/10] email_id=e001 extracted: topic=recruiter outreach, tone=positive
[2026-03-05 14:22:14] [WARNING] [2/10] email_id=e002 retrying after error: JSONDecodeError
[2026-03-05 14:22:16] [ERROR] [2/10] email_id=e002 failed after retry, saved to failed_extractions
[2026-03-05 14:22:45] [INFO] Job 1 completed in 34.2s: 9 success, 1 failed

El archivo de registro rota a los 5 MB y mantiene los últimos 3 archivos (worker.log, worker.log.1, worker.log.2).

Qué aprender del código

mcp_server/server.py

• FastMCP("email-insights") — crea la instancia del servidor con un nombre para mostrar

• @mcp.tool() — registra la función decorada como una herramienta MCP invocable

• Las cadenas de documentación (docstrings) importan — Claude las lee para decidir cuándo y cómo llamar a cada herramienta

• Sugerencias de tipo (Type hints) — FastMCP las utiliza para construir el esquema de entrada JSON que recibe Claude

• mcp.run() — inicia el bucle stdio; Claude Desktop se comunica a través de stdin/stdout

mcp_server/tools.py

• Completamente separado de MCP: funciones de Python simples que devuelven cadenas JSON

• Las consultas SQL parametrizadas evitan la inyección: WHERE topic LIKE ? con params

• La fábrica sqlite3.Row te permite acceder a las columnas por nombre: row["topic"]

• _ensure_jobs_tables() utiliza CREATE TABLE IF NOT EXISTS: es seguro llamarlo en cada invocación de herramienta

worker/job_runner.py

• Consulta SQLite cada 10 segundos: no se necesita un intermediario de mensajes, solo una base de datos compartida

• PRAGMA journal_mode=WAL permite que el servidor MCP lea mientras el trabajador escribe

• Lógica de reintento: un reintento en caso de tiempo de espera o JSON incorrecto, luego failed_extractions

• processed_emails se actualiza después de cada correo electrónico para que check_job_status_tool siempre refleje el progreso en vivo

utils/logger.py

• get_logger(name) es idempotente: es seguro llamarlo desde cualquier módulo, sin controladores duplicados

• RotatingFileHandler evita el crecimiento ilimitado del disco

• Utiliza sys.stderr para el controlador de flujo: sys.stdout está reservado para el protocolo JSON-RPC de MCP

ingestion/fetch_emails_imap.py

• imaplib.IMAP4_SSL — se conecta a cualquier servidor IMAP; credenciales cargadas desde .env

• mail.search(None, "ALL") devuelve todos los ID de mensaje; invertido para el orden de más reciente a más antiguo

• Las barras de progreso tqdm muestran el estado de obtención y almacenamiento en SQLite en vivo con el asunto actual como sufijo

• Almacena en la tabla raw_emails a través de db.raw_emails — idempotente (INSERT OR REPLACE)

• --output es opcional: el CSV solo se escribe cuando se pasa explícitamente

ingestion/extract_signals.py

• OpenAI(base_url="http://127.0.0.1:10101/v1") — apunta el cliente a LM Studio

• temperature=0.1 bajo — salida más determinista, mejor para JSON estructurado

• Elimina los delimitadores de código markdown que el LLM podría envolver alrededor de su respuesta JSON

• Vuelve a los valores predeterminados seguros si el análisis falla: la canalización nunca falla por un correo electrónico incorrecto