romtuck/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Websockets

Browse Source
This commit is contained in:
Haakam Aujla 2026-01-27 16:49:42 -08:00
parent 33871a8f99
commit cabba2ef91
11 changed files with 125 additions and 506 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
extensions/agentmail/CHANGELOG.md
View File

@ -6,7 +6,7 @@
- Initial AgentMail plugin release - Initial AgentMail plugin release
- Email channel integration via AgentMail API - Email channel integration via AgentMail API
- Webhook-based inbound message handling - WebSocket-based real-time message handling (no public URL required)
- Full thread context fetching for conversation history - Full thread context fetching for conversation history
- Sender allowFrom filtering (consistent with other channels) - Sender allowFrom filtering (consistent with other channels)
- Attachment metadata in thread context - Attachment metadata in thread context

9
extensions/agentmail/README.md
View File

@ -39,16 +39,9 @@ Or via config:
} }
``` ```
## Webhook Setup
Register a webhook in the AgentMail dashboard:
- **URL:** `https://your-gateway-host:port/webhooks/agentmail`
- **Event:** `message.received`
## Features ## Features
- Webhook-based inbound email handling - WebSocket-based real-time email handling (no public URL required)
- Full thread context for conversation history - Full thread context for conversation history
- Sender allowFrom filtering - Sender allowFrom filtering
- Attachment metadata with on-demand download URLs - Attachment metadata with on-demand download URLs

82
extensions/agentmail/src/accounts.test.ts
View File

@ -16,7 +16,6 @@ describe("resolveCredentials", () => {
const result = resolveCredentials(cfg, {}); const result = resolveCredentials(cfg, {});
expect(result.apiKey).toBe("am_config_token"); expect(result.apiKey).toBe("am_config_token");
expect(result.inboxId).toBe("inbox@agentmail.to"); expect(result.inboxId).toBe("inbox@agentmail.to");
expect(result.webhookPath).toBe("/webhooks/agentmail");
}); });
it("falls back to environment variables", () => { it("falls back to environment variables", () => {
@ -48,93 +47,12 @@ describe("resolveCredentials", () => {
expect(result.inboxId).toBe("config@agentmail.to"); expect(result.inboxId).toBe("config@agentmail.to");
}); });
it("resolves custom webhookPath from config", () => {
const cfg: CoreConfig = {
channels: {
agentmail: {
webhookPath: "/custom/path",
},
},
};
const result = resolveCredentials(cfg, {});
expect(result.webhookPath).toBe("/custom/path");
});
it("resolves webhookPath from env", () => {
const cfg: CoreConfig = {};
const env = {
AGENTMAIL_WEBHOOK_PATH: "/env/webhook",
};
const result = resolveCredentials(cfg, env);
expect(result.webhookPath).toBe("/env/webhook");
});
it("returns undefined for missing credentials", () => { it("returns undefined for missing credentials", () => {
const cfg: CoreConfig = {}; const cfg: CoreConfig = {};
const result = resolveCredentials(cfg, {}); const result = resolveCredentials(cfg, {});
expect(result.apiKey).toBeUndefined(); expect(result.apiKey).toBeUndefined();
expect(result.inboxId).toBeUndefined(); expect(result.inboxId).toBeUndefined();
}); });
it("resolves webhookUrl from config", () => {
const cfg: CoreConfig = {
channels: {
agentmail: {
webhookUrl: "https://my-gateway.ngrok.io",
},
},
};
const result = resolveCredentials(cfg, {});
expect(result.webhookUrl).toBe("https://my-gateway.ngrok.io");
});
it("resolves webhookUrl from env", () => {
const cfg: CoreConfig = {};
const env = {
AGENTMAIL_WEBHOOK_URL: "https://env-gateway.example.com/hooks/email",
};
const result = resolveCredentials(cfg, env);
expect(result.webhookUrl).toBe(
"https://env-gateway.example.com/hooks/email"
);
});
it("derives webhookPath from webhookUrl", () => {
const cfg: CoreConfig = {
channels: {
agentmail: {
webhookUrl: "https://my-gateway.ngrok.io/custom/path",
},
},
};
const result = resolveCredentials(cfg, {});
expect(result.webhookPath).toBe("/custom/path");
});
it("uses default webhookPath when URL has no path", () => {
const cfg: CoreConfig = {
channels: {
agentmail: {
webhookUrl: "https://my-gateway.ngrok.io",
},
},
};
const result = resolveCredentials(cfg, {});
expect(result.webhookPath).toBe("/webhooks/agentmail");
});
it("explicit webhookPath takes precedence over URL-derived path", () => {
const cfg: CoreConfig = {
channels: {
agentmail: {
webhookUrl: "https://my-gateway.ngrok.io/url-path",
webhookPath: "/explicit-path",
},
},
};
const result = resolveCredentials(cfg, {});
expect(result.webhookPath).toBe("/explicit-path");
});
}); });
describe("resolveAgentMailAccount", () => { describe("resolveAgentMailAccount", () => {

32
extensions/agentmail/src/accounts.ts
View File

@ -6,54 +6,24 @@ import type {
ResolvedAgentMailAccount, ResolvedAgentMailAccount,
} from "./utils.js"; } from "./utils.js";
/** Resolved AgentMail credentials and paths. */ /** Resolved AgentMail credentials. */
export type ResolvedAgentMailCredentials = { export type ResolvedAgentMailCredentials = {
apiKey?: string; apiKey?: string;
inboxId?: string; inboxId?: string;
webhookUrl?: string;
webhookPath: string;
}; };
export const DEFAULT_WEBHOOK_PATH = "/webhooks/agentmail";
/** Extracts the path from a URL string. Returns undefined if just root "/". */
export function extractPathFromUrl(url: string): string | undefined {
try {
const parsed = new URL(url);
// Return undefined if just root path "/" - use default instead
if (!parsed.pathname || parsed.pathname === "/") {
return undefined;
}
return parsed.pathname;
} catch {
return undefined;
}
}
/** /**
* Resolves AgentMail credentials from config and environment. * Resolves AgentMail credentials from config and environment.
* Maps user-facing keys (token, emailAddress) to SDK names (apiKey, inboxId). * Maps user-facing keys (token, emailAddress) to SDK names (apiKey, inboxId).
* Derives webhookPath from webhookUrl if not explicitly set.
*/ */
export function resolveCredentials( export function resolveCredentials(
cfg: CoreConfig, cfg: CoreConfig,
env: Record<string, string | undefined> = process.env env: Record<string, string | undefined> = process.env
): ResolvedAgentMailCredentials { ): ResolvedAgentMailCredentials {
const base = cfg.channels?.agentmail ?? {}; const base = cfg.channels?.agentmail ?? {};
const webhookUrl = base.webhookUrl || env.AGENTMAIL_WEBHOOK_URL;
// Derive path from URL if not explicitly set
let webhookPath = base.webhookPath || env.AGENTMAIL_WEBHOOK_PATH;
if (!webhookPath && webhookUrl) {
webhookPath = extractPathFromUrl(webhookUrl);
}
webhookPath = webhookPath || DEFAULT_WEBHOOK_PATH;
return { return {
apiKey: base.token || env.AGENTMAIL_TOKEN, apiKey: base.token || env.AGENTMAIL_TOKEN,
inboxId: base.emailAddress || env.AGENTMAIL_EMAIL_ADDRESS, inboxId: base.emailAddress || env.AGENTMAIL_EMAIL_ADDRESS,
webhookUrl,
webhookPath,
}; };
} }

22
extensions/agentmail/src/channel.ts
View File

@ -59,14 +59,7 @@ export const agentmailPlugin: ChannelPlugin<ResolvedAgentMailAccount> = {
cfg: cfg as CoreConfig, cfg: cfg as CoreConfig,
sectionKey: "agentmail", sectionKey: "agentmail",
accountId, accountId,
clearBaseFields: [ clearBaseFields: ["name", "token", "emailAddress", "allowFrom"],
"name",
"token",
"emailAddress",
"webhookUrl",
"webhookPath",
"allowFrom",
],
}), }),
isConfigured: (account) => account.configured, isConfigured: (account) => account.configured,
describeAccount: (account) => ({ describeAccount: (account) => ({
@ -136,18 +129,9 @@ export const agentmailPlugin: ChannelPlugin<ResolvedAgentMailAccount> = {
const updates: Record<string, unknown> = { enabled: true }; const updates: Record<string, unknown> = { enabled: true };
if (!input.useEnv) { if (!input.useEnv) {
if (input.token?.trim()) updates.token = input.token.trim(); if (input.token?.trim()) updates.token = input.token.trim();
if (input.emailAddress?.trim()) if (input.emailAddress?.trim()) updates.emailAddress = input.emailAddress.trim();
updates.emailAddress = input.emailAddress.trim();
if (input.webhookPath?.trim())
updates.webhookPath = input.webhookPath.trim();
} }
return { return { ...cfg, channels: { ...(cfg as CoreConfig).channels, agentmail: { ...existing, ...updates } } };
...cfg,
channels: {
...(cfg as CoreConfig).channels,
agentmail: { ...existing, ...updates },
},
};
}, },
}, },

4
extensions/agentmail/src/config-schema.ts
View File

@ -13,10 +13,6 @@ export const AgentMailConfigSchema = z.object({
token: z.string().optional(), token: z.string().optional(),
/** AgentMail inbox email address to monitor (required). */ /** AgentMail inbox email address to monitor (required). */
emailAddress: z.string().optional(), emailAddress: z.string().optional(),
/** Full public webhook URL (e.g., https://my-gateway.ngrok.io/webhooks/agentmail). */
webhookUrl: z.string().optional(),
/** Local webhook path (default: /webhooks/agentmail). Derived from webhookUrl if not set. */
webhookPath: z.string().optional(),
/** Allowed sender emails/domains. Empty = allow all. */ /** Allowed sender emails/domains. Empty = allow all. */
allowFrom: z.array(z.string()).optional(), allowFrom: z.array(z.string()).optional(),
}); });

253
extensions/agentmail/src/monitor.ts
View File

@ -1,6 +1,4 @@
import type { IncomingMessage, ServerResponse } from "node:http"; import type { AgentMail } from "agentmail";
import { registerPluginHttpRoute } from "clawdbot/plugin-sdk";
import { resolveCredentials } from "./accounts.js"; import { resolveCredentials } from "./accounts.js";
import { getAgentMailClient } from "./client.js"; import { getAgentMailClient } from "./client.js";
@ -8,7 +6,7 @@ import { getAgentMailRuntime } from "./runtime.js";
import { checkSenderAllowed, labelMessageAllowed } from "./filtering.js"; import { checkSenderAllowed, labelMessageAllowed } from "./filtering.js";
import { extractMessageBody, fetchFormattedThread } from "./thread.js"; import { extractMessageBody, fetchFormattedThread } from "./thread.js";
import { parseEmailFromAddress, parseNameFromAddress } from "./utils.js"; import { parseEmailFromAddress, parseNameFromAddress } from "./utils.js";
import type { CoreConfig, MessageReceivedEvent } from "./utils.js"; import type { CoreConfig } from "./utils.js";
export type MonitorAgentMailOptions = { export type MonitorAgentMailOptions = {
accountId?: string | null; accountId?: string | null;
@ -16,28 +14,13 @@ export type MonitorAgentMailOptions = {
}; };
// Runtime state tracking // Runtime state tracking
type RuntimeState = { type RuntimeState = { running: boolean; lastStartAt: number | null; lastStopAt: number | null; lastError: string | null; lastInboundAt?: number | null; lastOutboundAt?: number | null };
running: boolean;
lastStartAt: number | null;
lastStopAt: number | null;
lastError: string | null;
lastInboundAt?: number | null;
lastOutboundAt?: number | null;
};
const runtimeState = new Map<string, RuntimeState>(); const runtimeState = new Map<string, RuntimeState>();
const defaultState: RuntimeState = { const defaultState: RuntimeState = { running: false, lastStartAt: null, lastStopAt: null, lastError: null };
running: false,
lastStartAt: null,
lastStopAt: null,
lastError: null,
};
function recordState(accountId: string, state: Partial<RuntimeState>) { function recordState(accountId: string, state: Partial<RuntimeState>) {
const key = `agentmail:${accountId}`; const key = `agentmail:${accountId}`;
runtimeState.set(key, { runtimeState.set(key, { ...(runtimeState.get(key) ?? defaultState), ...state });
...(runtimeState.get(key) ?? defaultState),
...state,
});
} }
/** Returns runtime state for status checks. */ /** Returns runtime state for status checks. */
@ -45,28 +28,14 @@ export function getAgentMailRuntimeState(accountId: string) {
return runtimeState.get(`agentmail:${accountId}`); return runtimeState.get(`agentmail:${accountId}`);
} }
// HTTP helpers /** Builds message body from event as fallback when thread fetch fails. */
async function readBody(req: IncomingMessage): Promise<string> { function buildFallbackBody(message: AgentMail.messages.Message): string {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk);
return Buffer.concat(chunks).toString("utf-8");
}
function sendJson(res: ServerResponse, status: number, data: unknown) {
res.writeHead(status, { "Content-Type": "application/json" });
res.end(JSON.stringify(data));
}
/** Builds message body from webhook payload as fallback when API fetch fails. */
function buildFallbackBody(
message: NonNullable<MessageReceivedEvent["message"]>
): string {
const subject = message.subject ? `Subject: ${message.subject}\n\n` : ""; const subject = message.subject ? `Subject: ${message.subject}\n\n` : "";
return `${subject}${extractMessageBody(message)}`; return `${subject}${extractMessageBody(message)}`;
} }
/** /**
* Main monitor function that sets up webhook handling for AgentMail. * Main monitor function that sets up WebSocket connection to AgentMail.
*/ */
export async function monitorAgentMailProvider( export async function monitorAgentMailProvider(
opts: MonitorAgentMailOptions = {} opts: MonitorAgentMailOptions = {}
@ -79,15 +48,13 @@ export async function monitorAgentMailProvider(
return; return;
} }
const logger = core.logging.getChildLogger({ const logger = core.logging.getChildLogger({ module: "agentmail-monitor" });
module: "agentmail-auto-reply",
});
const logVerbose = (msg: string) => { const logVerbose = (msg: string) => {
if (core.logging.shouldLogVerbose()) (logger.debug ?? logger.info)(msg); if (core.logging.shouldLogVerbose()) (logger.debug ?? logger.info)(msg);
}; };
const accountId = opts.accountId ?? "default"; const accountId = opts.accountId ?? "default";
const { apiKey, inboxId, webhookPath } = resolveCredentials(cfg); const { apiKey, inboxId } = resolveCredentials(cfg);
if (!apiKey || !inboxId) { if (!apiKey || !inboxId) {
logger.warn("AgentMail not configured (missing token or email address)"); logger.warn("AgentMail not configured (missing token or email address)");
return; return;
@ -96,52 +63,58 @@ export async function monitorAgentMailProvider(
const client = getAgentMailClient(apiKey); const client = getAgentMailClient(apiKey);
const allowFrom = agentmailConfig?.allowFrom ?? []; const allowFrom = agentmailConfig?.allowFrom ?? [];
recordState(accountId, { recordState(accountId, { running: true, lastStartAt: Date.now(), lastError: null });
running: true, logger.info(`AgentMail: connecting WebSocket for ${inboxId}`);
lastStartAt: Date.now(),
lastError: null,
});
logger.info(`AgentMail: starting monitor for ${inboxId}`);
/** let socket: Awaited<ReturnType<typeof client.websockets.connect>> | null = null;
* Handles incoming webhook requests from AgentMail. let connectionCount = 0;
*/
const handleWebhook = async (req: IncomingMessage, res: ServerResponse) => { const subscribe = () => {
try { socket?.sendSubscribe({
if (req.method !== "POST") { type: "subscribe",
res.writeHead(405, { "Content-Type": "text/plain" }); inboxIds: [inboxId],
res.end("Method Not Allowed"); eventTypes: ["message.received"],
});
};
try {
socket = await client.websockets.connect();
socket.on("open", () => {
connectionCount++;
const isReconnect = connectionCount > 1;
logger.info(`AgentMail: WebSocket ${isReconnect ? "reconnected" : "connected"}, subscribing to ${inboxId}`);
subscribe();
if (isReconnect) {
recordState(accountId, { lastError: null }); // Clear error on successful reconnect
}
});
socket.on("message", async (event) => {
if (event.type === "subscribed") {
const sub = event as AgentMail.Subscribed;
logger.info(`AgentMail: subscribed to ${sub.inboxIds?.join(", ") ?? "inbox"}`);
return; return;
} }
const body = await readBody(req);
const payload = JSON.parse(body) as { eventType?: string };
// Only handle message.received events // Only handle message.received events
if (payload.eventType !== "message.received") { if (event.type !== "event") return;
return sendJson(res, 200, { ok: true, ignored: true }); const msgEvent = event as AgentMail.MessageReceivedEvent;
} if (msgEvent.eventType !== "message.received") return;
const event = payload as MessageReceivedEvent; const message = msgEvent.message;
const message = event.message; if (!message) return;
if (!message) {
return sendJson(res, 200, { ok: true, ignored: true });
}
// Parse sender email from "from" string (format: "Display Name <email>" or "email")
const senderEmail = parseEmailFromAddress(message.from); const senderEmail = parseEmailFromAddress(message.from);
logVerbose(`agentmail: received message from ${senderEmail}`); logVerbose(`agentmail: received message from ${senderEmail}`);
// Apply sender filtering // Apply sender filtering
const allowed = checkSenderAllowed(senderEmail, { allowFrom }); if (!checkSenderAllowed(senderEmail, { allowFrom })) {
if (!allowed) {
logVerbose(`agentmail: sender ${senderEmail} not in allowFrom`); logVerbose(`agentmail: sender ${senderEmail} not in allowFrom`);
return sendJson(res, 200, { ok: true, filtered: true }); return;
} }
// Label message as allowed (best effort - don't fail if labeling fails) // Label message as allowed (best effort)
try { try {
await labelMessageAllowed(client, inboxId, message.messageId); await labelMessageAllowed(client, inboxId, message.messageId);
} catch (labelErr) { } catch (labelErr) {
@ -150,45 +123,25 @@ export async function monitorAgentMailProvider(
recordState(accountId, { lastInboundAt: Date.now() }); recordState(accountId, { lastInboundAt: Date.now() });
// Fetch the full thread from API (handles webhook size limits for large messages) // Fetch the full thread from API
const threadBody = await fetchFormattedThread( const threadBody = await fetchFormattedThread(client, inboxId, message.threadId);
client,
inboxId,
message.threadId
);
// Extract current message text for RawBody/CommandBody
const messageBody = extractMessageBody(message); const messageBody = extractMessageBody(message);
// Fall back to webhook payload if thread fetch fails
const fullBody = threadBody || buildFallbackBody(message); const fullBody = threadBody || buildFallbackBody(message);
// Resolve routing // Resolve routing
const route = core.channel.routing.resolveAgentRoute({ const route = core.channel.routing.resolveAgentRoute({
cfg, cfg,
channel: "agentmail", channel: "agentmail",
peer: { peer: { kind: "dm", id: senderEmail },
kind: "dm",
id: senderEmail,
},
}); });
const senderName = parseNameFromAddress(message.from); const senderName = parseNameFromAddress(message.from);
const timestamp = new Date(message.timestamp).getTime(); const timestamp = new Date(message.timestamp).getTime();
// Format envelope // Format envelope
const storePath = core.channel.session.resolveStorePath( const storePath = core.channel.session.resolveStorePath(cfg.session?.store, { agentId: route.agentId });
cfg.session?.store, const envelopeOptions = core.channel.reply.resolveEnvelopeFormatOptions(cfg);
{ const previousTimestamp = core.channel.session.readSessionUpdatedAt({ storePath, sessionKey: route.sessionKey });
agentId: route.agentId,
}
);
const envelopeOptions =
core.channel.reply.resolveEnvelopeFormatOptions(cfg);
const previousTimestamp = core.channel.session.readSessionUpdatedAt({
storePath,
sessionKey: route.sessionKey,
});
const formattedBody = core.channel.reply.formatAgentEnvelope({ const formattedBody = core.channel.reply.formatAgentEnvelope({
channel: "Email", channel: "Email",
from: senderName, from: senderName,
@ -234,9 +187,7 @@ export async function monitorAgentMailProvider(
to: inboxId, to: inboxId,
accountId: route.accountId, accountId: route.accountId,
}, },
onRecordError: (err) => { onRecordError: (err) => logger.warn(`Failed updating session meta: ${String(err)}`),
logger.warn(`Failed updating session meta: ${String(err)}`);
},
}); });
const preview = messageBody.slice(0, 200).replace(/\n/g, "\\n"); const preview = messageBody.slice(0, 200).replace(/\n/g, "\\n");
@ -244,80 +195,64 @@ export async function monitorAgentMailProvider(
const { dispatcher, replyOptions, markDispatchIdle } = const { dispatcher, replyOptions, markDispatchIdle } =
core.channel.reply.createReplyDispatcherWithTyping({ core.channel.reply.createReplyDispatcherWithTyping({
humanDelay: core.channel.reply.resolveHumanDelayConfig( humanDelay: core.channel.reply.resolveHumanDelayConfig(cfg, route.agentId),
cfg,
route.agentId
),
deliver: async (payload) => { deliver: async (payload) => {
// Import outbound dynamically to avoid circular deps
const { sendAgentMailReply } = await import("./outbound.js"); const { sendAgentMailReply } = await import("./outbound.js");
const text = payload.text ?? ""; const text = payload.text ?? "";
if (!text) return; // Skip empty replies if (!text) return;
await sendAgentMailReply({ await sendAgentMailReply({ client, inboxId, messageId: message.messageId, text });
client,
inboxId,
messageId: message.messageId,
text,
});
recordState(accountId, { lastOutboundAt: Date.now() }); recordState(accountId, { lastOutboundAt: Date.now() });
}, },
onError: (err, info) => { onError: (err, info) => logger.error(`agentmail ${info.kind} reply failed: ${String(err)}`),
logger.error(`agentmail ${info.kind} reply failed: ${String(err)}`);
},
}); });
const { queuedFinal, counts } = const { queuedFinal, counts } = await core.channel.reply.dispatchReplyFromConfig({
await core.channel.reply.dispatchReplyFromConfig({ ctx: ctxPayload,
ctx: ctxPayload, cfg,
cfg, dispatcher,
dispatcher, replyOptions,
replyOptions, });
});
markDispatchIdle(); markDispatchIdle();
if (queuedFinal) { if (queuedFinal) {
logVerbose( logVerbose(`agentmail: delivered ${counts.final} reply(ies) to ${senderEmail}`);
`agentmail: delivered ${counts.final} reply(ies) to ${senderEmail}`
);
core.system.enqueueSystemEvent(`Email from ${senderName}: ${preview}`, { core.system.enqueueSystemEvent(`Email from ${senderName}: ${preview}`, {
sessionKey: route.sessionKey, sessionKey: route.sessionKey,
contextKey: `agentmail:message:${message.messageId}`, contextKey: `agentmail:message:${message.messageId}`,
}); });
} }
});
sendJson(res, 200, { ok: true }); socket.on("error", (error) => {
} catch (err) { logger.error(`AgentMail WebSocket error: ${String(error)}`);
logger.error(`agentmail webhook handler failed: ${String(err)}`); recordState(accountId, { lastError: String(error) });
sendJson(res, 500, { ok: false, error: String(err) }); });
}
};
// Register the webhook route socket.on("close", (event) => {
const unregisterHttp = registerPluginHttpRoute({ // SDK's ReconnectingWebSocket will auto-reconnect (default 30 attempts)
path: webhookPath, // On reconnect, "open" fires again and we resubscribe
pluginId: "agentmail", logger.warn(`AgentMail: WebSocket closed (code: ${event.code}), will attempt reconnect`);
accountId, });
log: logVerbose,
handler: handleWebhook,
});
logger.info(`AgentMail: webhook registered at ${webhookPath}`); // Wait for abort signal
await new Promise<void>((resolve) => {
const onAbort = () => {
logVerbose("agentmail: stopping monitor");
socket?.close();
recordState(accountId, { running: false, lastStopAt: Date.now() });
resolve();
};
// Wait for abort signal if (opts.abortSignal?.aborted) {
await new Promise<void>((resolve) => { onAbort();
const onAbort = () => { return;
logVerbose("agentmail: stopping monitor"); }
unregisterHttp();
recordState(accountId, { running: false, lastStopAt: Date.now() });
resolve();
};
if (opts.abortSignal?.aborted) { opts.abortSignal?.addEventListener("abort", onAbort, { once: true });
onAbort(); });
return; } catch (err) {
} logger.error(`AgentMail WebSocket connection failed: ${String(err)}`);
recordState(accountId, { running: false, lastError: String(err), lastStopAt: Date.now() });
opts.abortSignal?.addEventListener("abort", onAbort, { once: true }); }
});
} }

30
extensions/agentmail/src/onboarding.test.ts
View File

@ -87,34 +87,4 @@ describe("updateAgentMailConfig", () => {
"example.org", "example.org",
]); ]);
}); });
it("sets webhookPath", () => {
const cfg = {} as never;
const result: AnyConfig = updateAgentMailConfig(cfg, {
webhookPath: "/custom/webhook",
});
expect(result.channels?.agentmail?.webhookPath).toBe("/custom/webhook");
});
it("sets webhookUrl", () => {
const cfg = {} as never;
const result: AnyConfig = updateAgentMailConfig(cfg, {
webhookUrl: "https://my-gateway.ngrok.io",
});
expect(result.channels?.agentmail?.webhookUrl).toBe(
"https://my-gateway.ngrok.io"
);
});
it("sets both webhookUrl and webhookPath", () => {
const cfg = {} as never;
const result: AnyConfig = updateAgentMailConfig(cfg, {
webhookUrl: "https://gateway.example.com",
webhookPath: "/hooks/email",
});
expect(result.channels?.agentmail?.webhookUrl).toBe(
"https://gateway.example.com"
);
expect(result.channels?.agentmail?.webhookPath).toBe("/hooks/email");
});
}); });

192
extensions/agentmail/src/onboarding.ts
View File

@ -5,22 +5,14 @@ import type {
} from "clawdbot/plugin-sdk"; } from "clawdbot/plugin-sdk";
import { DEFAULT_ACCOUNT_ID, normalizeAccountId } from "clawdbot/plugin-sdk"; import { DEFAULT_ACCOUNT_ID, normalizeAccountId } from "clawdbot/plugin-sdk";
import { import { resolveAgentMailAccount, resolveCredentials } from "./accounts.js";
DEFAULT_WEBHOOK_PATH,
extractPathFromUrl,
resolveAgentMailAccount,
resolveCredentials,
} from "./accounts.js";
import type { AgentMailConfig, CoreConfig } from "./utils.js"; import type { AgentMailConfig, CoreConfig } from "./utils.js";
const channel = "agentmail" as const; const channel = "agentmail" as const;
const DEFAULT_DOMAIN = "agentmail.to"; const DEFAULT_DOMAIN = "agentmail.to";
/** Parses input into username and domain. Supports "user" or "user@domain". */ /** Parses input into username and domain. Supports "user" or "user@domain". */
export function parseInboxInput(input: string): { export function parseInboxInput(input: string): { username: string; domain: string } {
username: string;
domain: string;
} {
const trimmed = input.trim().toLowerCase(); const trimmed = input.trim().toLowerCase();
if (trimmed.includes("@")) { if (trimmed.includes("@")) {
const [username, domain] = trimmed.split("@"); const [username, domain] = trimmed.split("@");
@ -36,12 +28,8 @@ async function createInbox(
domain: string, domain: string,
displayName?: string displayName?: string
): Promise<string> { ): Promise<string> {
const inbox = await client.inboxes.create({ const inbox = await client.inboxes.create({ username, domain, displayName });
username, return inbox.inboxId;
domain,
displayName,
});
return inbox.inboxId; // inboxId is the email address
} }
/** Lists existing inboxes for the user. */ /** Lists existing inboxes for the user. */
@ -59,13 +47,7 @@ export function updateAgentMailConfig(
const agentmail = (channels.agentmail ?? {}) as AgentMailConfig; const agentmail = (channels.agentmail ?? {}) as AgentMailConfig;
return { return {
...cfg, ...cfg,
channels: { channels: { ...channels, agentmail: { ...agentmail, ...updates } },
...channels,
agentmail: {
...agentmail,
...updates,
},
},
} as MoltbotConfig; } as MoltbotConfig;
} }
@ -78,9 +60,7 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
return { return {
channel, channel,
configured, configured,
statusLines: [ statusLines: [`AgentMail: ${configured ? `configured (${inboxId})` : "needs token"}`],
`AgentMail: ${configured ? `configured (${inboxId})` : "needs token"}`,
],
selectionHint: configured ? "configured" : "not configured", selectionHint: configured ? "configured" : "not configured",
quickstartScore: configured ? 1 : 5, quickstartScore: configured ? 1 : 5,
}; };
@ -92,16 +72,11 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
: DEFAULT_ACCOUNT_ID; : DEFAULT_ACCOUNT_ID;
let next = cfg as MoltbotConfig; let next = cfg as MoltbotConfig;
const account = resolveAgentMailAccount({ const account = resolveAgentMailAccount({ cfg: next as CoreConfig, accountId });
cfg: next as CoreConfig,
accountId,
});
const { apiKey, inboxId } = resolveCredentials(next as CoreConfig); const { apiKey, inboxId } = resolveCredentials(next as CoreConfig);
const configured = Boolean(apiKey && inboxId); const configured = Boolean(apiKey && inboxId);
const canUseEnv = const canUseEnv = accountId === DEFAULT_ACCOUNT_ID && Boolean(process.env.AGENTMAIL_TOKEN?.trim());
accountId === DEFAULT_ACCOUNT_ID &&
Boolean(process.env.AGENTMAIL_TOKEN?.trim());
// If env var token is available and not already configured, offer to use it // If env var token is available and not already configured, offer to use it
if (canUseEnv && !account.configured) { if (canUseEnv && !account.configured) {
@ -110,13 +85,9 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
initialValue: true, initialValue: true,
}); });
if (useEnv) { if (useEnv) {
const envToken = process.env.AGENTMAIL_TOKEN!.trim(); const client = new AgentMailClient({ apiKey: process.env.AGENTMAIL_TOKEN!.trim() });
const client = new AgentMailClient({ apiKey: envToken });
const emailAddress = await selectOrCreateInbox(client, prompter); const emailAddress = await selectOrCreateInbox(client, prompter);
return { cfg: updateAgentMailConfig(next, { enabled: true, emailAddress }) };
return {
cfg: updateAgentMailConfig(next, { enabled: true, emailAddress }),
};
} }
} }
@ -126,9 +97,7 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
message: `AgentMail already configured (${inboxId}). Keep current settings?`, message: `AgentMail already configured (${inboxId}). Keep current settings?`,
initialValue: true, initialValue: true,
}); });
if (keep) { if (keep) return { cfg: next };
return { cfg: next };
}
} }
// Show help // Show help
@ -137,7 +106,7 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
"AgentMail is free to use at https://agentmail.to", "AgentMail is free to use at https://agentmail.to",
"", "",
"Get your API token from the dashboard, then we'll help you", "Get your API token from the dashboard, then we'll help you",
"create an inbox.", "create an inbox. No webhooks needed - we use WebSockets!",
].join("\n"), ].join("\n"),
"AgentMail Setup" "AgentMail Setup"
); );
@ -158,64 +127,6 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
// Apply config // Apply config
next = updateAgentMailConfig(next, { enabled: true, token, emailAddress }); next = updateAgentMailConfig(next, { enabled: true, token, emailAddress });
// Ask if gateway has a public URL
const hasPublicUrl = await prompter.confirm({
message:
"Does your gateway have a public URL? (for automatic webhook setup)",
initialValue: false,
});
let webhookUrl: string | undefined;
let webhookPath = DEFAULT_WEBHOOK_PATH;
if (hasPublicUrl) {
webhookUrl = String(
await prompter.text({
message: "Full webhook URL",
placeholder: `https://my-gateway.ngrok.io${DEFAULT_WEBHOOK_PATH}`,
validate: (v) => {
const trimmed = v?.trim();
if (!trimmed) return "Required";
if (
!trimmed.startsWith("http://") &&
!trimmed.startsWith("https://")
)
return "Must start with http:// or https://";
return undefined;
},
})
).trim();
webhookPath = extractPathFromUrl(webhookUrl) ?? DEFAULT_WEBHOOK_PATH;
// Auto-register webhook with AgentMail
try {
await client.webhooks.create({
url: webhookUrl,
eventTypes: ["message.received"],
clientId: `clawdbot-${emailAddress}`, // Idempotent per inbox
});
await prompter.note(
`Webhook registered: ${webhookUrl}`,
"Webhook Created"
);
} catch (err) {
// Webhook may already exist or API error - show warning but continue
await prompter.note(
[
`Could not auto-register webhook: ${String(err)}`,
"",
"You may need to configure it manually in the AgentMail dashboard:",
` URL: ${webhookUrl}`,
" Event: message.received",
].join("\n"),
"Webhook Warning"
);
}
}
// Save webhook config (webhookPath derived from URL or default)
next = updateAgentMailConfig(next, { webhookUrl, webhookPath });
// Ask about allowFrom // Ask about allowFrom
const addAllowFrom = await prompter.confirm({ const addAllowFrom = await prompter.confirm({
message: "Add senders to allowFrom? (Empty = allow all)", message: "Add senders to allowFrom? (Empty = allow all)",
@ -225,47 +136,23 @@ export const agentmailOnboardingAdapter: ChannelOnboardingAdapter = {
if (addAllowFrom) { if (addAllowFrom) {
const entry = String( const entry = String(
await prompter.text({ await prompter.text({
message: message: "Email or domain to allow (e.g., user@example.com or example.com)",
"Email or domain to allow (e.g., user@example.com or example.com)",
}) })
).trim(); ).trim();
if (entry) { if (entry) {
const existing = const existing = (next as CoreConfig).channels?.agentmail?.allowFrom ?? [];
(next as CoreConfig).channels?.agentmail?.allowFrom ?? [];
next = updateAgentMailConfig(next, { allowFrom: [...existing, entry] }); next = updateAgentMailConfig(next, { allowFrom: [...existing, entry] });
} }
} }
// Show manual webhook instructions only if we didn't auto-register
if (!webhookUrl) {
await prompter.note(
[
"Configure the webhook in your AgentMail dashboard:",
` URL: https://your-gateway${webhookPath}`,
" Event: message.received",
"",
"The gateway must be publicly accessible for webhooks to work.",
"Without this, Moltbot won't receive incoming emails.",
].join("\n"),
"Webhook Setup Required"
);
}
return { cfg: next }; return { cfg: next };
}, },
}; };
type Prompter = Parameters< type Prompter = Parameters<ChannelOnboardingAdapter["configure"]>[0]["prompter"];
ChannelOnboardingAdapter["configure"]
>[0]["prompter"];
/** Prompts user to select an existing inbox or create a new one. */ /** Prompts user to select an existing inbox or create a new one. */
async function selectOrCreateInbox( async function selectOrCreateInbox(client: AgentMailClient, prompter: Prompter): Promise<string> {
client: AgentMailClient,
prompter: Prompter
): Promise<string> {
// Check for existing inboxes
let existingInboxes: string[] = []; let existingInboxes: string[] = [];
try { try {
existingInboxes = await listInboxes(client); existingInboxes = await listInboxes(client);
@ -278,26 +165,15 @@ async function selectOrCreateInbox(
...existingInboxes.map((email) => ({ value: email, label: email })), ...existingInboxes.map((email) => ({ value: email, label: email })),
{ value: "__create__", label: "Create a new inbox" }, { value: "__create__", label: "Create a new inbox" },
]; ];
const selection = await prompter.select({ message: "Select an inbox or create a new one", options: choices });
const selection = await prompter.select({ if (selection !== "__create__") return selection as string;
message: "Select an inbox or create a new one",
options: choices,
});
if (selection !== "__create__") {
return selection as string;
}
} }
return promptForNewInbox(client, prompter); return promptForNewInbox(client, prompter);
} }
/** Prompts for inbox address and creates it, with retry on conflict. */ /** Prompts for inbox address and creates it, with retry on conflict. */
async function promptForNewInbox( async function promptForNewInbox(client: AgentMailClient, prompter: Prompter): Promise<string> {
client: AgentMailClient,
prompter: Prompter
): Promise<string> {
// eslint-disable-next-line no-constant-condition
while (true) { while (true) {
const input = String( const input = String(
await prompter.text({ await prompter.text({
@ -306,9 +182,8 @@ async function promptForNewInbox(
validate: (v) => { validate: (v) => {
if (!v?.trim()) return "Required"; if (!v?.trim()) return "Required";
const { username } = parseInboxInput(v); const { username } = parseInboxInput(v);
if (!/^[a-z0-9][a-z0-9._-]*[a-z0-9]$|^[a-z0-9]$/.test(username)) { if (!/^[a-z0-9][a-z0-9._-]*[a-z0-9]$|^[a-z0-9]$/.test(username))
return "Username must use lowercase letters, numbers, dots, underscores, or hyphens"; return "Username must use lowercase letters, numbers, dots, underscores, or hyphens";
}
return undefined; return undefined;
}, },
}) })
@ -317,36 +192,17 @@ async function promptForNewInbox(
const { username, domain } = parseInboxInput(input); const { username, domain } = parseInboxInput(input);
const targetEmail = `${username}@${domain}`; const targetEmail = `${username}@${domain}`;
const displayName = const displayName = String(await prompter.text({ message: "Display name (optional)", placeholder: "My Agent" })).trim() || undefined;
String(
await prompter.text({
message: "Display name (optional)",
placeholder: "My Agent",
})
).trim() || undefined;
try { try {
const emailAddress = await createInbox( const emailAddress = await createInbox(client, username, domain, displayName);
client,
username,
domain,
displayName
);
await prompter.note(`Your new inbox: ${emailAddress}`, "Inbox Created"); await prompter.note(`Your new inbox: ${emailAddress}`, "Inbox Created");
return emailAddress; return emailAddress;
} catch (err) { } catch (err) {
const msg = err instanceof Error ? err.message : String(err); const msg = err instanceof Error ? err.message : String(err);
const lower = msg.toLowerCase(); const lower = msg.toLowerCase();
if ( if (lower.includes("already") || lower.includes("taken") || lower.includes("exists") || msg.includes("409")) {
lower.includes("already") || await prompter.note(`${targetEmail} is already taken. Please try a different address.`, "Address Unavailable");
lower.includes("taken") ||
lower.includes("exists") ||
msg.includes("409")
) {
await prompter.note(
`${targetEmail} is already taken. Please try a different address.`,
"Address Unavailable"
);
continue; continue;
} }
throw new Error(`Failed to create inbox: ${msg}`); throw new Error(`Failed to create inbox: ${msg}`);

2
extensions/agentmail/src/thread.ts
View File

@ -72,7 +72,7 @@ export async function fetchFormattedThread(
return `${header}\n\n${messages}`; return `${header}\n\n${messages}`;
} catch { } catch {
// Caller handles fallback to webhook payload // Caller handles fallback to event message
return ""; return "";
} }
} }

3
extensions/agentmail/src/utils.ts
View File

@ -31,9 +31,6 @@ export type ResolvedAgentMailAccount = {
/** SDK Message type alias. */ /** SDK Message type alias. */
export type Message = AgentMail.messages.Message; export type Message = AgentMail.messages.Message;
/** Re-export SDK webhook event type. */
export type MessageReceivedEvent = AgentMail.events.MessageReceivedEvent;
/** Formats a date as a UTC string. */ /** Formats a date as a UTC string. */
export function formatUtcDate(date: Date | string | number): string { export function formatUtcDate(date: Date | string | number): string {
return new Date(date).toUTCString().replace("GMT", "UTC"); return new Date(date).toUTCString().replace("GMT", "UTC");