Daniele
0c04ba2e4c
WhatsApp delivers a history sync at login, not on reconnect, so an in-memory store left the connector blind after every process restart (23 in two weeks). The store is now mirrored to store/, next to auth/ and bind-mounted the same way. Not SQLite: node:sqlite needs Node 22 (unflagged only from 24) and the runtime image ships Debian trixie's nodejs = 20.19.2; better-sqlite3 is native and the slim image has no toolchain. So an append-only JSONL log for messages plus a debounced JSON snapshot for chats/contacts, both written tmp+rename so a crash cannot truncate them. Compaction on load and every 500 appends keeps the log from creeping upward. Also fixes a pre-existing duplication bug: pushMessage appended unconditionally, re-adding every message a history re-sync redelivered. It now returns false on a known id and the message is skipped in both the transcript and the log. MAX_MSGS_PER_CHAT 200 -> 500. logout deletes store/ with auth/, so re-linking a different phone cannot inherit the previous account's history. Data at rest: message text is now written to the user's bind-mounted home. Nothing but session keys was persisted before. Verified on skald-runtime:v4 with a seeded store: 751 lines with 50 duplicates, a 700-message chat and a torn trailing line -> 701 loaded, compacted to 501, cap applied, second run 501 -> 501 unchanged.
781 lines
33 KiB
JavaScript
781 lines
33 KiB
JavaScript
#!/usr/bin/env node
|
||
|
||
/**
|
||
* WhatsApp MCP Server (JSON-RPC 2.0 over stdio) — Baileys edition.
|
||
*
|
||
* Runs INSIDE the user's per-user container (blueprint §6/§7). Unlike the old
|
||
* whatsapp-web.js server, this one uses `@whiskeysockets/baileys`: a pure-WebSocket
|
||
* WhatsApp multi-device client with **no browser** — so it fits the slim
|
||
* `skald-runtime` image (node, no Chromium) and needs no puppeteer self-healing.
|
||
*
|
||
* ── Interactive login contract (the generic §15 seam) ───────────────────────────
|
||
* A per-user connector that needs an interactive login exposes ONE standard tool,
|
||
* `login_status`, that Skald's login API calls directly (never the agent). It
|
||
* returns a small JSON object the login panel renders:
|
||
*
|
||
* { "state": "connecting" | "need_scan" | "ready" | "logged_out",
|
||
* "qr": "data:image/png;base64,…" // present only while state == need_scan
|
||
* "message": "human-readable line" }
|
||
*
|
||
* The panel polls it; when `state == "ready"` Skald flips the connector's
|
||
* `auth_state` to `ready`. WhatsApp's credential is the persisted session on disk
|
||
* (`./auth/`, under the bind-mounted home → survives a container recreate), not a
|
||
* token — so there is nothing to paste back, only a QR to scan.
|
||
*
|
||
* ── Why this file is ESM ────────────────────────────────────────────────────────
|
||
* Baileys 7.x ships ESM-only (`"type": "module"`, engines node >= 20). Importing it
|
||
* from CommonJS would rely on Node's `require(esm)` bridge, which is a moving target
|
||
* across Node versions — so the connector is ESM too, like the other node connectors.
|
||
*/
|
||
|
||
import fs from 'node:fs';
|
||
import path from 'node:path';
|
||
import readline from 'node:readline';
|
||
import nodeCrypto from 'node:crypto';
|
||
import { fileURLToPath } from 'node:url';
|
||
import qrcode from 'qrcode';
|
||
|
||
// Baileys uses the Web Crypto global (`crypto.subtle`). Node exposes it as
|
||
// `globalThis.crypto` from v19+, but keep the polyfill so an older runtime still
|
||
// reaches the QR instead of dying on connect with "crypto is not defined".
|
||
if (!globalThis.crypto) globalThis.crypto = nodeCrypto.webcrypto;
|
||
|
||
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
||
|
||
// ── Paths ──────────────────────────────────────────────────────────────────
|
||
// Everything hangs off __dirname (the connector dir inside the container home,
|
||
// `~/.skald/mcp/<name>/`), which is bind-mounted and therefore durable.
|
||
const AUTH_DIR = path.join(__dirname, 'auth'); // multi-file auth state (the "session")
|
||
const MEDIA_DIR = path.join(__dirname, 'media');
|
||
|
||
// ── Logging ────────────────────────────────────────────────────────────────────
|
||
// Skald funnels every user's whatsapp stderr into ONE shared log file, so each line
|
||
// carries the linked account once we know it — otherwise two users' failures are
|
||
// indistinguishable after the fact.
|
||
let logTag = 'whatsapp_mcp';
|
||
function log(msg) { process.stderr.write(`[${logTag}] ${msg}\n`); }
|
||
|
||
// A silent logger: Baileys requires one, and anything it prints must never reach
|
||
// stdout (that channel is reserved for JSON-RPC framing).
|
||
const silentLogger = (() => {
|
||
const noop = () => {};
|
||
const l = { level: 'silent', trace: noop, debug: noop, info: noop, warn: noop, error: noop, fatal: noop };
|
||
l.child = () => l;
|
||
return l;
|
||
})();
|
||
|
||
// `libsignal` bypasses the Baileys logger and writes straight to the console: one
|
||
// "Failed to decrypt…" line plus a full stack trace *per candidate session* for every
|
||
// message it cannot open. That turned ~30 real failures into 1400+ log lines. Collapse
|
||
// the burst into a single counted line, and force any stray console.log to stderr —
|
||
// stdout belongs to the JSON-RPC framing and a dependency printing there corrupts it.
|
||
const rawConsoleError = console.error.bind(console);
|
||
let decryptFailures = 0;
|
||
console.error = (...args) => {
|
||
const first = typeof args[0] === 'string' ? args[0] : '';
|
||
if (first.startsWith('Session error:')) return; // per-session stack spam
|
||
if (first.startsWith('Failed to decrypt message with any known session')) {
|
||
decryptFailures++;
|
||
log(`could not decrypt an incoming message (${decryptFailures} since start)`);
|
||
return;
|
||
}
|
||
rawConsoleError(...args);
|
||
};
|
||
console.log = (...args) => console.error(...args);
|
||
console.info = console.warn = console.error;
|
||
|
||
// ── Baileys (loaded dynamically so a missing install degrades gracefully) ───────
|
||
let makeWASocket, useMultiFileAuthState, DisconnectReason, fetchLatestBaileysVersion,
|
||
jidNormalizedUser, WAMessageStubType;
|
||
|
||
async function loadBaileys() {
|
||
try {
|
||
const baileys = await import('@whiskeysockets/baileys');
|
||
makeWASocket = baileys.default || baileys.makeWASocket;
|
||
useMultiFileAuthState = baileys.useMultiFileAuthState;
|
||
DisconnectReason = baileys.DisconnectReason;
|
||
fetchLatestBaileysVersion = baileys.fetchLatestBaileysVersion;
|
||
jidNormalizedUser = baileys.jidNormalizedUser;
|
||
WAMessageStubType = baileys.WAMessageStubType;
|
||
return true;
|
||
} catch (e) {
|
||
log(`FATAL: baileys not installed (${e.message}). Run npm install.`);
|
||
return false;
|
||
}
|
||
}
|
||
|
||
// ── Connection state ─────────────────────────────────────────────────────────
|
||
// connecting – socket starting or reconnecting
|
||
// need_scan – a QR is available; the user must scan it
|
||
// ready – authenticated and connected; tools operational
|
||
// logged_out – the phone unlinked this device; a fresh QR + scan is required
|
||
let state = 'connecting';
|
||
let sock = null;
|
||
let curQr = null; // latest raw QR string (null once scanned / connected)
|
||
let meJid = null;
|
||
let starting = false;
|
||
|
||
// Every socket gets a generation number. Events arriving from a superseded socket are
|
||
// dropped, so a late `close` from a dead socket can never schedule a second reconnect
|
||
// or stomp on the live one's state.
|
||
let generation = 0;
|
||
let reconnectTimer = null;
|
||
|
||
const RECONNECT_BASE_MS = 1500;
|
||
const RECONNECT_MAX_MS = 60_000;
|
||
let backoffMs = RECONNECT_BASE_MS;
|
||
|
||
// ── Store ─────────────────────────────────────────────────────────────────────
|
||
// Baileys keeps no chat/contact store of its own; we build a minimal one from the
|
||
// history-sync event and live upserts. It is mirrored to `store/` (bind-mounted next
|
||
// to `auth/`, so it survives a container recreate) because WhatsApp only delivers a
|
||
// history sync at *login* — without a mirror every process restart left the connector
|
||
// blind until new messages happened to arrive.
|
||
//
|
||
// Deliberately not SQLite: `node:sqlite` needs Node 22 (the runtime image is 20.19.2)
|
||
// and `better-sqlite3` is a native module the slim image cannot build. So: an
|
||
// append-only JSONL log for messages (O(1) per message) plus a small debounced
|
||
// snapshot for chats/contacts.
|
||
const chats = new Map(); // jid -> { id, name, unread, conversationTimestamp }
|
||
const contacts = new Map(); // jid -> { id, name }
|
||
const messages = new Map(); // jid -> [ { id, fromMe, ts, text, author } ] (capped)
|
||
|
||
const MAX_MSGS_PER_CHAT = 500;
|
||
|
||
const STORE_DIR = path.join(__dirname, 'store');
|
||
const MSG_LOG = path.join(STORE_DIR, 'messages.jsonl');
|
||
const META_FILE = path.join(STORE_DIR, 'meta.json');
|
||
const COMPACT_EVERY = 500; // appends between rewrites of the log
|
||
const META_DEBOUNCE_MS = 5000;
|
||
|
||
let appendsSinceCompact = 0;
|
||
let metaTimer = null;
|
||
|
||
// Returns false when the message was already known: history sync re-delivers the same
|
||
// ids on every login, and without this both the log and the transcript would grow a
|
||
// duplicate per restart.
|
||
function pushMessage(jid, m) {
|
||
if (!jid) return false;
|
||
let arr = messages.get(jid);
|
||
if (!arr) { arr = []; messages.set(jid, arr); }
|
||
if (m.id && arr.some((x) => x.id === m.id)) return false;
|
||
arr.push(m);
|
||
if (arr.length > MAX_MSGS_PER_CHAT) arr.splice(0, arr.length - MAX_MSGS_PER_CHAT);
|
||
return true;
|
||
}
|
||
|
||
function writeFileAtomic(file, data) {
|
||
const tmp = `${file}.tmp`;
|
||
fs.writeFileSync(tmp, data);
|
||
fs.renameSync(tmp, file); // rename is atomic: a crash mid-write cannot truncate the store
|
||
}
|
||
|
||
function loadStore() {
|
||
try {
|
||
const meta = JSON.parse(fs.readFileSync(META_FILE, 'utf8'));
|
||
for (const c of meta.chats || []) if (c?.id) chats.set(c.id, c);
|
||
for (const c of meta.contacts || []) if (c?.id) contacts.set(c.id, c);
|
||
} catch { /* first run, or a snapshot we cannot parse — start empty */ }
|
||
|
||
let restored = 0;
|
||
try {
|
||
for (const line of fs.readFileSync(MSG_LOG, 'utf8').split('\n')) {
|
||
if (!line) continue;
|
||
try {
|
||
const { jid, m } = JSON.parse(line);
|
||
if (pushMessage(jid, m)) restored++;
|
||
} catch { /* skip a torn trailing line */ }
|
||
}
|
||
} catch { /* no log yet */ }
|
||
|
||
if (restored || chats.size || contacts.size) {
|
||
log(`store loaded: ${chats.size} chats, ${contacts.size} contacts, ${restored} messages`);
|
||
}
|
||
// Rewrite immediately: replays the per-chat cap and drops duplicates, so the log
|
||
// cannot creep upward across restarts.
|
||
if (restored) compactMessageLog();
|
||
}
|
||
|
||
function compactMessageLog() {
|
||
try {
|
||
fs.mkdirSync(STORE_DIR, { recursive: true });
|
||
const out = [];
|
||
for (const [jid, arr] of messages) for (const m of arr) out.push(JSON.stringify({ jid, m }));
|
||
writeFileAtomic(MSG_LOG, out.length ? out.join('\n') + '\n' : '');
|
||
appendsSinceCompact = 0;
|
||
} catch (e) {
|
||
log(`store compaction failed: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
function appendMessage(jid, m) {
|
||
try {
|
||
fs.mkdirSync(STORE_DIR, { recursive: true });
|
||
fs.appendFileSync(MSG_LOG, JSON.stringify({ jid, m }) + '\n');
|
||
if (++appendsSinceCompact >= COMPACT_EVERY) compactMessageLog();
|
||
} catch (e) {
|
||
log(`store append failed: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
function saveMeta() {
|
||
try {
|
||
fs.mkdirSync(STORE_DIR, { recursive: true });
|
||
writeFileAtomic(META_FILE, JSON.stringify({
|
||
chats: [...chats.values()],
|
||
contacts: [...contacts.values()],
|
||
}));
|
||
} catch (e) {
|
||
log(`store meta save failed: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
// chats/contacts churn in bursts during a history sync; one write per burst is enough.
|
||
function scheduleMetaSave() {
|
||
if (metaTimer) return;
|
||
metaTimer = setTimeout(() => { metaTimer = null; saveMeta(); }, META_DEBOUNCE_MS);
|
||
}
|
||
|
||
function flushStore() {
|
||
if (metaTimer) { clearTimeout(metaTimer); metaTimer = null; }
|
||
saveMeta();
|
||
}
|
||
|
||
function clearStore() {
|
||
try { fs.rmSync(STORE_DIR, { recursive: true, force: true }); } catch { /* nothing to clear */ }
|
||
appendsSinceCompact = 0;
|
||
if (metaTimer) { clearTimeout(metaTimer); metaTimer = null; }
|
||
}
|
||
|
||
function contactName(jid) {
|
||
const c = contacts.get(jid);
|
||
if (c && c.name) return c.name;
|
||
const ch = chats.get(jid);
|
||
if (ch && ch.name) return ch.name;
|
||
return jid ? jid.split('@')[0] : 'unknown';
|
||
}
|
||
|
||
function textOf(msg) {
|
||
const m = msg.message;
|
||
if (!m) return '';
|
||
return (
|
||
m.conversation ||
|
||
m.extendedTextMessage?.text ||
|
||
m.imageMessage?.caption ||
|
||
m.videoMessage?.caption ||
|
||
m.documentMessage?.caption ||
|
||
(m.imageMessage ? '[image]' : '') ||
|
||
(m.videoMessage ? '[video]' : '') ||
|
||
(m.audioMessage ? '[audio]' : '') ||
|
||
(m.documentMessage ? '[document]' : '') ||
|
||
(m.stickerMessage ? '[sticker]' : '') ||
|
||
''
|
||
);
|
||
}
|
||
|
||
// ── WhatsApp protocol version (cached) ─────────────────────────────────────────
|
||
// `fetchLatestBaileysVersion` is a network call. Doing it on every reconnect meant
|
||
// ~15 outbound requests a day, each able to hand us a protocol version the installed
|
||
// library cannot speak. Fetch once, reuse for a few hours, fall back to the last good
|
||
// value (or the library default) when the fetch fails.
|
||
const VERSION_TTL_MS = 6 * 60 * 60 * 1000;
|
||
let cachedVersion = null;
|
||
let cachedVersionAt = 0;
|
||
|
||
async function getWAVersion() {
|
||
const now = Date.now();
|
||
if (cachedVersion && now - cachedVersionAt < VERSION_TTL_MS) return cachedVersion;
|
||
try {
|
||
const { version } = await fetchLatestBaileysVersion();
|
||
cachedVersion = version;
|
||
cachedVersionAt = now;
|
||
log(`WhatsApp protocol version ${version.join('.')}`);
|
||
} catch (e) {
|
||
log(`version fetch failed (${e.message}) — using ${cachedVersion ? 'cached value' : 'library default'}`);
|
||
}
|
||
return cachedVersion || undefined;
|
||
}
|
||
|
||
// ── WhatsApp socket lifecycle ──────────────────────────────────────────────────
|
||
|
||
// Detach and close a socket for good. Baileys keeps its own keepalive and, more
|
||
// importantly, a `creds.update → saveCreds` handler bound to the auth-state snapshot
|
||
// it was built with. Leaving an old socket alive meant two writers over the same
|
||
// `auth/` directory on every reconnect — the way Signal sessions end up inconsistent
|
||
// and messages start failing with "Bad MAC".
|
||
function teardownSock(s) {
|
||
if (!s) return;
|
||
try { s.ev.removeAllListeners(); } catch { /* already gone */ }
|
||
try { s.ws?.removeAllListeners?.(); } catch { /* already gone */ }
|
||
try { s.end(undefined); } catch { /* already closed */ }
|
||
}
|
||
|
||
function scheduleReconnect(delayMs) {
|
||
if (reconnectTimer) return; // one pending reconnect at a time
|
||
reconnectTimer = setTimeout(() => {
|
||
reconnectTimer = null;
|
||
startSock().catch((e) => log(`reconnect failed: ${e.message}`));
|
||
}, delayMs);
|
||
}
|
||
|
||
function disconnectName(code) {
|
||
const name = DisconnectReason?.[code];
|
||
return typeof name === 'string' ? name : 'unknown';
|
||
}
|
||
|
||
async function startSock() {
|
||
if (starting) { scheduleReconnect(2000); return; }
|
||
starting = true;
|
||
const myGen = ++generation;
|
||
try {
|
||
if (!makeWASocket) { state = 'connecting'; return; }
|
||
|
||
// Retire the previous socket BEFORE building a new one.
|
||
teardownSock(sock);
|
||
sock = null;
|
||
|
||
fs.mkdirSync(AUTH_DIR, { recursive: true });
|
||
const { state: authState, saveCreds } = await useMultiFileAuthState(AUTH_DIR);
|
||
const version = await getWAVersion();
|
||
|
||
const s = makeWASocket({
|
||
version,
|
||
auth: authState,
|
||
logger: silentLogger,
|
||
// The browser identity is not cosmetic: `getWebInfo` only asks the phone for a
|
||
// desktop-grade history sync when browser[0] is 'Mac OS' or 'Windows'. With any
|
||
// other name the sub-platform stays WEB_BROWSER and `syncFullHistory` is inert.
|
||
// Trade-off: the phone lists this device as "Mac OS Chrome", not "Skald".
|
||
browser: ['Mac OS', 'Chrome', '121.0.0'],
|
||
syncFullHistory: true,
|
||
// Set explicitly rather than left to the default. Baileys 6.7.x derives this from
|
||
// `syncFullHistory` and would otherwise gate history processing off entirely;
|
||
// 6.17.x and 7.x default it to true. Pinning it here makes the behaviour the same
|
||
// whichever version is installed.
|
||
shouldSyncHistoryMessage: () => true,
|
||
markOnlineOnConnect: false,
|
||
generateHighQualityLinkPreview: false,
|
||
});
|
||
sock = s;
|
||
|
||
s.ev.on('creds.update', saveCreds);
|
||
|
||
s.ev.on('connection.update', (u) => {
|
||
if (myGen !== generation) return; // event from a superseded socket
|
||
const { connection, lastDisconnect, qr } = u;
|
||
if (qr) { curQr = qr; state = 'need_scan'; log('QR ready — awaiting scan'); }
|
||
if (connection === 'open') {
|
||
curQr = null;
|
||
state = 'ready';
|
||
backoffMs = RECONNECT_BASE_MS;
|
||
meJid = s.user?.id ? jidNormalizedUser(s.user.id) : null;
|
||
const who = s.user?.name || (meJid ? meJid.split('@')[0] : null);
|
||
if (who) logTag = `whatsapp_mcp ${who}`;
|
||
log('connection open — ready');
|
||
}
|
||
if (connection === 'close') {
|
||
const code = lastDisconnect?.error?.output?.statusCode;
|
||
if (code === DisconnectReason.loggedOut) {
|
||
state = 'logged_out';
|
||
curQr = null;
|
||
log('logged out by phone — clearing session');
|
||
teardownSock(s);
|
||
sock = null;
|
||
try { fs.rmSync(AUTH_DIR, { recursive: true, force: true }); } catch { /* nothing to clear */ }
|
||
starting = false;
|
||
scheduleReconnect(500); // produce a fresh QR immediately
|
||
} else {
|
||
state = 'connecting';
|
||
log(`connection closed (${code ?? '?'} ${disconnectName(code)}) — reconnecting in ${Math.round(backoffMs / 1000)}s`);
|
||
teardownSock(s);
|
||
sock = null;
|
||
starting = false;
|
||
scheduleReconnect(backoffMs);
|
||
backoffMs = Math.min(backoffMs * 2, RECONNECT_MAX_MS);
|
||
}
|
||
}
|
||
});
|
||
|
||
// Initial history sync: chats, contacts and a batch of messages.
|
||
s.ev.on('messaging-history.set', ({ chats: hc, contacts: hcs, messages: hm, syncType, progress }) => {
|
||
if (myGen !== generation) return;
|
||
for (const c of hc || []) {
|
||
chats.set(c.id, {
|
||
id: c.id,
|
||
name: c.name || c.subject || null,
|
||
unread: c.unreadCount || 0,
|
||
conversationTimestamp: Number(c.conversationTimestamp) || 0,
|
||
});
|
||
}
|
||
for (const c of hcs || []) {
|
||
contacts.set(c.id, { id: c.id, name: c.name || c.notify || c.verifiedName || null });
|
||
}
|
||
for (const m of hm || []) ingestMessage(m, false);
|
||
scheduleMetaSave();
|
||
log(`history sync (type ${syncType ?? '?'}${progress != null ? `, ${progress}%` : ''}): ` +
|
||
`+${(hc || []).length} chats, +${(hcs || []).length} contacts, +${(hm || []).length} messages ` +
|
||
`→ ${chats.size} chats / ${contacts.size} contacts known`);
|
||
});
|
||
|
||
s.ev.on('chats.upsert', (cs) => {
|
||
if (myGen !== generation) return;
|
||
for (const c of cs) chats.set(c.id, {
|
||
id: c.id, name: c.name || c.subject || null,
|
||
unread: c.unreadCount || 0,
|
||
conversationTimestamp: Number(c.conversationTimestamp) || 0,
|
||
});
|
||
scheduleMetaSave();
|
||
});
|
||
s.ev.on('contacts.upsert', (cs) => {
|
||
if (myGen !== generation) return;
|
||
for (const c of cs) contacts.set(c.id, { id: c.id, name: c.name || c.notify || c.verifiedName || null });
|
||
scheduleMetaSave();
|
||
});
|
||
s.ev.on('contacts.update', (cs) => {
|
||
if (myGen !== generation) return;
|
||
for (const c of cs) {
|
||
const prev = contacts.get(c.id) || { id: c.id };
|
||
contacts.set(c.id, { ...prev, name: c.name || c.notify || prev.name || null });
|
||
}
|
||
scheduleMetaSave();
|
||
});
|
||
|
||
s.ev.on('messages.upsert', ({ messages: ms, type }) => {
|
||
if (myGen !== generation) return;
|
||
for (const m of ms) ingestMessage(m, type === 'notify');
|
||
});
|
||
} catch (e) {
|
||
log(`startSock error: ${e.message}`);
|
||
state = 'connecting';
|
||
scheduleReconnect(backoffMs);
|
||
backoffMs = Math.min(backoffMs * 2, RECONNECT_MAX_MS);
|
||
} finally {
|
||
starting = false;
|
||
}
|
||
}
|
||
|
||
function ingestMessage(m, live) {
|
||
try {
|
||
const jid = m.key?.remoteJid;
|
||
if (!jid || jid === 'status@broadcast') return;
|
||
|
||
// A message Signal could not open arrives with no `message` payload and the
|
||
// CIPHERTEXT stub type. It used to be stored with an empty text, so `get_messages`
|
||
// rendered it as a blank line and the agent had no way to tell a silent gap from a
|
||
// genuinely empty message. Keep it, but say what it is.
|
||
const undecryptable = WAMessageStubType != null &&
|
||
m.messageStubType === WAMessageStubType.CIPHERTEXT;
|
||
const text = undecryptable ? '[undecryptable message]' : textOf(m);
|
||
|
||
// Protocol/system frames (reactions, receipts, key distribution…) carry no text
|
||
// and are pure noise in a transcript.
|
||
if (!text) return;
|
||
|
||
const stored = {
|
||
id: m.key?.id,
|
||
fromMe: !!m.key?.fromMe,
|
||
ts: Number(m.messageTimestamp) || 0,
|
||
text,
|
||
author: m.key?.participant || (m.key?.fromMe ? meJid : jid),
|
||
};
|
||
if (!pushMessage(jid, stored)) return; // already known — do not log it twice
|
||
appendMessage(jid, stored);
|
||
|
||
if (live && !chats.has(jid)) {
|
||
chats.set(jid, { id: jid, name: m.pushName || null, unread: 0, conversationTimestamp: Number(m.messageTimestamp) || 0 });
|
||
scheduleMetaSave();
|
||
} else if (live) {
|
||
const ch = chats.get(jid);
|
||
ch.conversationTimestamp = Number(m.messageTimestamp) || ch.conversationTimestamp;
|
||
if (m.pushName && !ch.name) ch.name = m.pushName;
|
||
scheduleMetaSave();
|
||
}
|
||
} catch { /* one malformed frame must not stop the stream */ }
|
||
}
|
||
|
||
// ── Helpers ────────────────────────────────────────────────────────────────────
|
||
|
||
// Turn a plain phone number or a chat id into a WhatsApp jid.
|
||
function toJid(chat_id, number) {
|
||
if (chat_id && chat_id.includes('@')) return chat_id;
|
||
const raw = (chat_id || number || '').replace(/[^0-9]/g, '');
|
||
if (!raw) return null;
|
||
return `${raw}@s.whatsapp.net`;
|
||
}
|
||
|
||
function requireReady() {
|
||
if (state !== 'ready') {
|
||
throw new Error(`WhatsApp is not connected (state: ${state}). ` +
|
||
(state === 'need_scan' || state === 'logged_out'
|
||
? 'Open the connector in Skald and scan the QR code to sign in.'
|
||
: 'It is still connecting — try again in a few seconds.'));
|
||
}
|
||
}
|
||
|
||
// ── Tools: interactive login (the §15 generic contract) ─────────────────────────
|
||
|
||
async function toolLoginStatus() {
|
||
let qrDataUrl = null;
|
||
if (state === 'need_scan' && curQr) {
|
||
try { qrDataUrl = await qrcode.toDataURL(curQr, { width: 320, margin: 2 }); } catch { /* no QR to render */ }
|
||
}
|
||
const message = {
|
||
connecting: 'Connecting to WhatsApp…',
|
||
need_scan: 'Scan this QR code: WhatsApp → Settings → Linked Devices → Link a Device.',
|
||
ready: 'WhatsApp is connected.',
|
||
logged_out: 'This device was unlinked. Scan the new QR code to sign in again.',
|
||
}[state] || state;
|
||
// Returned as a JSON string in a text content part; the login API parses it.
|
||
return JSON.stringify({ state, qr: qrDataUrl, message });
|
||
}
|
||
|
||
async function toolStatus() {
|
||
const s = await toolLoginStatus();
|
||
const { state: st, message } = JSON.parse(s);
|
||
return `WhatsApp status: ${st.toUpperCase()}\n${message}` +
|
||
(st === 'ready'
|
||
? `\nKnown chats: ${chats.size}` +
|
||
`\nKnown contacts: ${contacts.size}` +
|
||
(decryptFailures ? `\nUndecryptable messages since start: ${decryptFailures}` : '')
|
||
: '');
|
||
}
|
||
|
||
async function toolLogout() {
|
||
try { if (sock) await sock.logout(); } catch { /* already gone */ }
|
||
generation++; // orphan any in-flight socket events
|
||
teardownSock(sock);
|
||
sock = null;
|
||
if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
|
||
try { fs.rmSync(AUTH_DIR, { recursive: true, force: true }); } catch { /* nothing to clear */ }
|
||
chats.clear(); contacts.clear(); messages.clear();
|
||
clearStore();
|
||
curQr = null; state = 'connecting'; starting = false; meJid = null;
|
||
backoffMs = RECONNECT_BASE_MS;
|
||
scheduleReconnect(500);
|
||
return 'Logged out and cleared the session. A new QR code will be generated — open the connector in Skald and scan it.';
|
||
}
|
||
|
||
// ── Tools: messaging ────────────────────────────────────────────────────────────
|
||
|
||
async function toolListChats(args) {
|
||
requireReady();
|
||
const max = Math.min(Math.max(1, args.max_chats || 20), 50);
|
||
const list = [...chats.values()]
|
||
.sort((a, b) => (b.conversationTimestamp || 0) - (a.conversationTimestamp || 0))
|
||
.slice(0, max);
|
||
if (!list.length) return 'No chats known yet. History may still be syncing — try again in a few seconds.';
|
||
const lines = [`Recent WhatsApp chats (${list.length}):`];
|
||
for (const c of list) {
|
||
const kind = c.id.endsWith('@g.us') ? '[group]' : '[chat]';
|
||
const unread = c.unread ? ` (${c.unread} unread)` : '';
|
||
lines.push(`- ${c.name || contactName(c.id)} ${kind}${unread} | ID: ${c.id}`);
|
||
}
|
||
return lines.join('\n');
|
||
}
|
||
|
||
async function toolGetMessages(args) {
|
||
requireReady();
|
||
const jid = toJid(args.chat_id, args.number);
|
||
if (!jid) return 'Error: provide chat_id or number.';
|
||
const limit = Math.min(Math.max(1, args.limit || 20), 100);
|
||
const offset = Math.max(0, args.offset || 0);
|
||
const arr = (messages.get(jid) || []).slice().sort((a, b) => (a.ts || 0) - (b.ts || 0));
|
||
if (!arr.length) return `No messages buffered for ${contactName(jid)} (${jid}). Only messages seen since sign-in are available.`;
|
||
const end = arr.length - offset;
|
||
const slice = arr.slice(Math.max(0, end - limit), Math.max(0, end));
|
||
const lines = [`Messages with ${contactName(jid)} (${jid}):`];
|
||
for (const m of slice) {
|
||
const who = m.fromMe ? 'me' : (jid.endsWith('@g.us') ? contactName(m.author) : contactName(jid));
|
||
const when = m.ts ? new Date(m.ts * 1000).toISOString().replace('T', ' ').slice(0, 16) : '';
|
||
lines.push(`[${when}] ${who}: ${m.text}`);
|
||
}
|
||
return lines.join('\n');
|
||
}
|
||
|
||
async function toolSendMessage(args) {
|
||
requireReady();
|
||
const jid = toJid(args.chat_id, args.number);
|
||
if (!jid) return 'Error: provide chat_id or number.';
|
||
if (!args.message) return 'Error: message is required.';
|
||
await sock.sendMessage(jid, { text: String(args.message) });
|
||
return `Message sent to ${contactName(jid)} (${jid}).`;
|
||
}
|
||
|
||
async function toolSearchContacts(args) {
|
||
requireReady();
|
||
const q = String(args.query || '').toLowerCase();
|
||
if (!q) return 'Error: query is required.';
|
||
const max = Math.min(Math.max(1, args.max_results || 20), 50);
|
||
const seen = new Set();
|
||
const out = [];
|
||
for (const c of contacts.values()) {
|
||
if (out.length >= max) break;
|
||
const name = c.name || '';
|
||
if (name.toLowerCase().includes(q) || c.id.includes(q)) {
|
||
if (seen.has(c.id)) continue;
|
||
seen.add(c.id);
|
||
out.push(`- ${name || contactName(c.id)} | ID: ${c.id}`);
|
||
}
|
||
}
|
||
if (!out.length) return `No contacts found matching "${args.query}".`;
|
||
return [`Contacts matching "${args.query}" (${out.length}):`, ...out].join('\n');
|
||
}
|
||
|
||
// ── MCP tool definitions ────────────────────────────────────────────────────────
|
||
|
||
const TOOLS = [
|
||
{
|
||
name: 'login_status',
|
||
title: 'Login Status',
|
||
description: 'Interactive-login status for this connector (used by the Skald login panel). Returns a JSON object {state, qr, message}: state is connecting|need_scan|ready|logged_out; qr is a data-URL PNG present only while a scan is needed. Safe to poll.',
|
||
inputSchema: { type: 'object', properties: {} },
|
||
},
|
||
{
|
||
name: 'status',
|
||
title: 'Status',
|
||
description: 'WhatsApp connection status as a short human-readable report. Call this first when another WhatsApp tool fails.',
|
||
inputSchema: { type: 'object', properties: {} },
|
||
},
|
||
{
|
||
name: 'logout',
|
||
title: 'Logout',
|
||
description: 'Log out of WhatsApp: end the session, clear the stored credentials, and generate a fresh QR code to link a (possibly different) phone. After calling, the user must scan the new QR in the Skald connector page.',
|
||
inputSchema: { type: 'object', properties: {} },
|
||
},
|
||
{
|
||
name: 'list_chats',
|
||
title: 'List Chats',
|
||
description: 'List recent WhatsApp chats (contacts and groups) with name, ID and unread count. Only chats seen since sign-in / history sync are known.',
|
||
inputSchema: {
|
||
type: 'object',
|
||
properties: { max_chats: { type: 'integer', description: 'Max chats to return (default 20, max 50).' } },
|
||
},
|
||
},
|
||
{
|
||
name: 'get_messages',
|
||
title: 'Get Messages',
|
||
description: 'Get buffered messages from a chat. Identify it with EITHER chat_id (from list_chats) OR a phone number with country code for an individual contact. Only messages seen since sign-in are available (no deep history).',
|
||
inputSchema: {
|
||
type: 'object',
|
||
properties: {
|
||
chat_id: { type: 'string', description: 'Chat ID, e.g. "39XXXXXXXXXX@s.whatsapp.net" or "…@g.us".' },
|
||
number: { type: 'string', description: 'Alternative to chat_id: phone number with country code (e.g. "393331234567"). Ignored if chat_id is given.' },
|
||
limit: { type: 'integer', description: 'Number of messages (default 20, max 100).' },
|
||
offset: { type: 'integer', description: 'Skip this many of the most recent messages (default 0).' },
|
||
},
|
||
},
|
||
},
|
||
{
|
||
name: 'send_message',
|
||
title: 'Send Message',
|
||
description: 'Send a WhatsApp text message. Identify the recipient with EITHER chat_id (from list_chats, use for groups) OR a phone number with country code for an individual contact.',
|
||
inputSchema: {
|
||
type: 'object',
|
||
properties: {
|
||
chat_id: { type: 'string', description: 'Chat ID to send to (use for groups).' },
|
||
number: { type: 'string', description: 'Alternative to chat_id: phone number with country code. Ignored if chat_id is given.' },
|
||
message: { type: 'string', description: 'The text to send.' },
|
||
},
|
||
required: ['message'],
|
||
},
|
||
},
|
||
{
|
||
name: 'search_contacts',
|
||
title: 'Search Contacts',
|
||
description: 'Search known WhatsApp contacts by name or number. Use to find a contact ID to message.',
|
||
inputSchema: {
|
||
type: 'object',
|
||
properties: {
|
||
query: { type: 'string', description: 'Name or partial name/number (case-insensitive).' },
|
||
max_results: { type: 'integer', description: 'Max contacts to return (default 20, max 50).' },
|
||
},
|
||
required: ['query'],
|
||
},
|
||
},
|
||
];
|
||
|
||
// ── JSON-RPC framing ─────────────────────────────────────────────────────────
|
||
|
||
function okResponse(id, result) { return JSON.stringify({ jsonrpc: '2.0', id, result }); }
|
||
function textResult(id, text, isError = false) {
|
||
const result = { content: [{ type: 'text', text }] };
|
||
if (isError) result.isError = true;
|
||
return JSON.stringify({ jsonrpc: '2.0', id, result });
|
||
}
|
||
|
||
async function handleRequest(msg) {
|
||
const { method, id, params } = msg;
|
||
|
||
if (method === 'initialize') {
|
||
return okResponse(id, {
|
||
protocolVersion: '2024-11-05',
|
||
capabilities: { tools: {} },
|
||
serverInfo: { name: 'whatsapp', version: '2.1.0' },
|
||
});
|
||
}
|
||
if (method === 'notifications/initialized') return null;
|
||
if (method === 'tools/list') return okResponse(id, { tools: TOOLS });
|
||
|
||
if (method === 'tools/call') {
|
||
const toolName = params?.name || '';
|
||
const toolArgs = params?.arguments || {};
|
||
let text;
|
||
try {
|
||
switch (toolName) {
|
||
case 'login_status': text = await toolLoginStatus(); break;
|
||
case 'status': text = await toolStatus(); break;
|
||
case 'logout': text = await toolLogout(); break;
|
||
case 'list_chats': text = await toolListChats(toolArgs); break;
|
||
case 'get_messages': text = await toolGetMessages(toolArgs); break;
|
||
case 'send_message': text = await toolSendMessage(toolArgs); break;
|
||
case 'search_contacts': text = await toolSearchContacts(toolArgs); break;
|
||
default:
|
||
return textResult(id, `Unknown tool: ${toolName}`, true);
|
||
}
|
||
} catch (e) {
|
||
log(`tool '${toolName}' error: ${e.message}`);
|
||
return textResult(id, `Error: ${e.message}`, true);
|
||
}
|
||
const isErr = typeof text === 'string' && text.startsWith('Error:');
|
||
return textResult(id, text, isErr);
|
||
}
|
||
|
||
return JSON.stringify({ jsonrpc: '2.0', id, error: { code: -32601, message: `Method not found: ${method}` } });
|
||
}
|
||
|
||
// ── Main ─────────────────────────────────────────────────────────────────────
|
||
|
||
async function main() {
|
||
log('Starting WhatsApp MCP server (Baileys)');
|
||
fs.mkdirSync(MEDIA_DIR, { recursive: true });
|
||
loadStore();
|
||
|
||
if (await loadBaileys()) {
|
||
startSock().catch((e) => log(`initial startSock failed: ${e.message}`));
|
||
}
|
||
|
||
const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity });
|
||
rl.on('line', async (line) => {
|
||
line = line.trim();
|
||
if (!line) return;
|
||
let msg;
|
||
try { msg = JSON.parse(line); } catch (e) { log(`bad JSON on stdin: ${e.message}`); return; }
|
||
const resp = await handleRequest(msg);
|
||
if (resp !== null) process.stdout.write(resp + '\n');
|
||
});
|
||
rl.on('close', () => { log('stdin closed, shutting down'); shutdown(0); });
|
||
|
||
process.on('SIGTERM', () => { log('SIGTERM'); shutdown(0); });
|
||
process.on('SIGINT', () => { log('SIGINT'); shutdown(0); });
|
||
}
|
||
|
||
function shutdown(code) {
|
||
generation++;
|
||
teardownSock(sock);
|
||
sock = null;
|
||
flushStore();
|
||
process.exit(code);
|
||
}
|
||
|
||
main().catch((e) => { log(`Fatal: ${e.message}`); process.exit(1); });
|