Email Insights

电子邮件见解

一个MCP服务器，向Claude Desktop公开电子邮件信号分析，并配备一个后台工作程序，用于定时异步提取作业和结构化日志记录。

项目结构

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

设置

1.安装依赖项

pip install -r requirements.txt

2.配置IMAP凭据

复制 .env.example 到 .env 并填写您的凭据：

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

对于Gmail，请在以下位置生成特定于应用程序的密码 myaccount.google.com/apppasswords.

3.将电子邮件导入SQLite

从收件箱中提取所有电子邮件并将其存储在 raw_emails 表：

python ingestion/fetch_emails_imap.py

进度条显示实时获取和存储状态。选项：

# 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.启动LM工作室

• 打开LM Studio并加载以下模型（Llama 3、Mistral等）的任何指令
• 启动本地服务器： 本地服务器→ 启动服务器
• 默认URL: http://127.0.0.1:10101
• 复制模型标识符字符串并将其粘贴到 ingestion/extract_signals.py 作为 LOCAL_MODEL

5.运行信号提取

python ingestion/store_signals.py

上面写着 data/emails.csv，将每封电子邮件发送给您当地的LLM进行信号提取， 并将结果存储在 database/signals.db.

6.启动后台工作程序

工人是一个独立的流程，用于轮询预定的采掘工作。在专用终端中运行它：

python worker/job_runner.py

工人将所有活动记录到 logs/worker.log 并发送到stderr。它每10秒轮询一次SQLite，并自动拾取任何待处理或到期的计划作业。

7.连接克劳德桌面

将此服务器添加到您的Claude Desktop配置中：

雨衣： ~/Library/Application Support/Claude/claude_desktop_config.json

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

重新启动克劳德桌面。你应该看看 email-insights 在工具列表中。

MCP工具

查询工具

作业调度工具

所有调度工具立即返回。提取在工作进程中异步运行。

schedule_extraction_tool 运行模式

建筑

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

MCP服务器和工作程序是 两个完全独立的过程 它们只共享SQLite数据库。MCP服务器从不等待提取完成——它创建一个作业记录并立即返回。工人拥有所有对 jobs 和 failed_extractions 表（状态更新、进度、失败）；MCP服务器仅读取作业状态。

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'))
);

两者 jobs 和 failed_extractions 首次使用时自动创建，无需手动迁移。

结构化日志记录

所有工人活动都写在 logs/worker.log （自动创建）和stderr。

日志格式：

[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

日志文件以5MB的速度旋转，并保留最后3个文件(worker.log, worker.log.1, worker.log.2).

从代码中学到什么

mcp_server/server.py

• FastMCP("email-insights") --使用显示名称创建服务器实例
• @mcp.tool() --将装饰函数注册为可调用的MCP工具
• 文档字符串很重要 --Claude阅读它们以决定何时以及如何调用每个工具
• 键入提示 --FastMCP使用它们来构建Claude接收的JSON输入模式
• mcp.run() --启动stdio循环；Claude Desktop通过stdin/stdout进行通信

mcp_server/tools.py

• 与MCP完全分离——返回JSON字符串的普通Python函数
• 参数化SQL查询可防止注入： WHERE topic LIKE ? 和 params
• sqlite3.Row factory允许您按名称访问列： row["topic"]
• _ensure_jobs_tables() 用途 CREATE TABLE IF NOT EXISTS --每次调用工具时都可以安全调用

worker/job_runner.py

• 每10秒轮询一次SQLite——不需要消息代理，只需要一个共享数据库
• PRAGMA journal_mode=WAL 允许MCP服务器在工作程序写入时进行读取
• 重试逻辑：超时或JSON错误时重试一次，然后 failed_extractions
• processed_emails 每次发送电子邮件后都会更新 check_job_status_tool 始终反映实时进展

utils/logger.py

• get_logger(name) 是幂等的——从任何模块调用都是安全的，没有重复的处理程序
• RotatingFileHandler 防止磁盘无限增长
• 用途 sys.stderr 对于流处理程序-- sys.stdout 为MCP的JSON-RPC协议保留

ingestion/fetch_emails_imap.py

• imaplib.IMAP4_SSL --连接到任何IMAP服务器；从加载的凭据 .env
• mail.search(None, "ALL") 返回所有消息ID；与最近的第一顺序相反
• tqdm 进度条显示实时获取和SQLite存储状态，以当前主题为后缀
• 商店到 raw_emails 桌子via db.raw_emails --幂等性(INSERT OR REPLACE)
• --output 可选：CSV仅在显式传递时写入

ingestion/extract_signals.py

• OpenAI(base_url="http://127.0.0.1:10101/v1") --将客户指向LM Studio
• 低 temperature=0.1 --更具确定性的输出，更适合结构化JSON
• 剥离markdown代码围栏，LLM可能会围绕其JSON响应进行包装
• 如果解析失败，则返回安全默认值——管道永远不会在一封错误的电子邮件上崩溃