diff --git a/extension/channel/ax-channel.ts b/extension/channel/ax-channel.ts index da4f171..94ccaeb 100644 --- a/extension/channel/ax-channel.ts +++ b/extension/channel/ax-channel.ts @@ -22,9 +22,9 @@ import type { IncomingMessage, ServerResponse } from "node:http"; import type { PluginRuntime } from "clawdbot/plugin-sdk"; -import type { AxDispatchPayload, AxDispatchResponse, DispatchSession } from "../lib/types.js"; +import type { AxDispatchPayload, AxDispatchResponse, DispatchSession, OutboundConfig } from "../lib/types.js"; import { loadAgentRegistry, getAgent, verifySignature, logRegisteredAgents } from "../lib/auth.js"; -import { sendProgressUpdate } from "../lib/api.js"; +import { sendProgressUpdate, callAxTool } from "../lib/api.js"; import { buildMissionBriefing } from "../lib/context.js"; // ─── Agent Event Tracking ─────────────────────────────────────────────────── @@ -568,18 +568,54 @@ export function getDispatchSession(sessionKey: string): DispatchSession | undefi return undefined; } +/** + * Resolve auth token for outbound delivery. + * Tries tokenFile first, then AX_ACCESS_TOKEN env var. + */ +async function resolveOutboundToken(outboundCfg: OutboundConfig | Record): Promise { + const cfg = outboundCfg || {}; + const tokenFile = (cfg as OutboundConfig).tokenFile; + + if (tokenFile) { + try { + const fs = await import("node:fs"); + const tokenPath = tokenFile.replace(/^~/, process.env.HOME || ""); + const tokenData = JSON.parse(fs.readFileSync(tokenPath, "utf-8")); + if (tokenData.access_token) return tokenData.access_token; + } catch { + // Fall through to env var + } + } + + return process.env.AX_ACCESS_TOKEN; +} + /** * Create the aX Platform channel plugin */ export function createAxChannel(config: { + outbound?: OutboundConfig; agents?: Array<{ id: string; secret: string; handle?: string; env?: string }>; backendUrl?: string; }) { const backendUrl = config.backendUrl || process.env.AX_BACKEND_URL || "http://localhost:8001"; + const outboundConfig = config.outbound || {}; // Load agent registry from config loadAgentRegistry(config.agents); + // Probe cache — one MCP call covers all prod accounts + let probeCache: { result: any; ts: number } | null = null; + + // Helper: get agents array from clawdbot config + function getPluginAgents(cfg: any): Array<{ id: string; secret: string; handle?: string; env?: string }> { + const pluginAgents = cfg?.plugins?.entries?.["ax-platform"]?.config?.agents; + if (Array.isArray(pluginAgents) && pluginAgents.length > 0) { + return pluginAgents; + } + return config.agents || []; + } + return { id: "ax-platform", meta: { @@ -593,21 +629,185 @@ export function createAxChannel(config: { capabilities: { chatTypes: ["direct", "group"], + media: { images: true, documents: true }, + threads: true, + mentions: true, }, config: { - listAccountIds: () => ["default"], - resolveAccount: () => ({ accountId: "default" }), + /** + * List all configured account IDs + */ + listAccountIds: (cfg: any) => { + return getPluginAgents(cfg).map((a) => a.handle?.replace(/^@/, "") || a.id); + }, + + /** + * Default account ID - first prod agent, or first agent + */ + defaultAccountId: (cfg: any) => { + const agents = getPluginAgents(cfg); + const prod = agents.find((a) => a.env === "prod"); + const agent = prod || agents[0]; + return agent ? (agent.handle?.replace(/^@/, "") || agent.id) : "default"; + }, + + /** + * Check if account is enabled + */ + isEnabled: (account: any) => account?.enabled !== false, + + /** + * Check if account has required credentials + */ + isConfigured: (account: any) => Boolean(account?.agentId && account?.secret), + + /** + * Describe account for dashboard listing + */ + describeAccount: (account: any) => ({ + accountId: account?.accountId || "default", + name: account?.handle ? `@${account.handle.replace(/^@/, "")}` : account?.accountId, + enabled: account?.enabled !== false, + configured: Boolean(account?.agentId && account?.secret), + }), + + /** + * Resolve account config by ID + */ + resolveAccount: (cfg: any, accountId?: string) => { + const agents = getPluginAgents(cfg); + const agent = agents.find((a) => + a.handle?.replace(/^@/, "") === accountId || + a.id === accountId + ) || (accountId ? undefined : agents[0]); + + if (agent) { + return { + accountId: agent.handle?.replace(/^@/, "") || agent.id, + agentId: agent.id, + secret: agent.secret, + handle: agent.handle, + enabled: true, + env: agent.env, + }; + } + + return { accountId: accountId || "default" }; + }, }, outbound: { - deliveryMode: "direct", + deliveryMode: "direct" as const, + chunker: null, + textChunkLimit: 4000, + + /** + * Send text to aX Platform (for heartbeat/cron delivery) + * Matches Clawdbot's standard channel outbound interface + */ + async sendText({ + to, + text, + accountId, + cfg, + replyToId, + threadId, + deps, + }: { + to: string; + text: string; + accountId?: string; + cfg?: any; + replyToId?: string; + threadId?: string; + deps?: any; + }) { + const logger = runtime?.logger || { info: console.log, error: console.error }; + + const authToken = await resolveOutboundToken(outboundConfig); + if (!authToken) { + logger.error("[ax-platform] Outbound sendText failed: No access token configured"); + return { channel: "ax-platform", ok: false, error: "No access token" }; + } + + const outboundCfg = outboundConfig || {}; + const mcpEndpoint = outboundCfg.mcpEndpoint || process.env.AX_MCP_ENDPOINT || "https://mcp.paxai.app"; + const spaceId = to?.replace(/^space:/, "") || outboundCfg.defaultSpaceId; + if (!spaceId) { + logger.error("[ax-platform] Outbound sendText failed: No target space"); + return { channel: "ax-platform", ok: false, error: "No target space" }; + } - // Handle agent responses - this is called by the dispatcher - async sendText({ text, sessionKey }: { text: string; sessionKey?: string }) { - // For aX, responses are returned via HTTP response in the dispatch handler - // This is a no-op since we use sync-over-async pattern - return { ok: true }; + try { + const result = await callAxTool(mcpEndpoint, authToken, "messages", { + action: "send", + content: text, + space_id: spaceId, + }) as Record; + + const preview = text.length > 50 ? text.substring(0, 50) + "..." : text; + logger.info(`[ax-platform] Outbound sent to ${spaceId}: ${preview}`); + return { channel: "ax-platform", ok: true, messageId: result.message_id }; + } catch (err) { + logger.error(`[ax-platform] Outbound sendText error: ${err}`); + return { channel: "ax-platform", ok: false, error: String(err) }; + } + }, + + /** + * Send media to aX Platform (for heartbeat/cron delivery) + * Matches Clawdbot's standard channel outbound interface + */ + async sendMedia({ + to, + text, + mediaUrl, + accountId, + cfg, + replyToId, + threadId, + deps, + }: { + to: string; + text?: string; + mediaUrl: string; + accountId?: string; + cfg?: any; + replyToId?: string; + threadId?: string; + deps?: any; + }) { + const logger = runtime?.logger || { info: console.log, error: console.error }; + + const authToken = await resolveOutboundToken(outboundConfig); + if (!authToken) { + logger.error("[ax-platform] Outbound sendMedia failed: No access token configured"); + return { channel: "ax-platform", ok: false, error: "No access token" }; + } + + const outboundCfg = outboundConfig || {}; + const mcpEndpoint = outboundCfg.mcpEndpoint || process.env.AX_MCP_ENDPOINT || "https://mcp.paxai.app"; + const spaceId = to?.replace(/^space:/, "") || outboundCfg.defaultSpaceId; + if (!spaceId) { + logger.error("[ax-platform] Outbound sendMedia failed: No target space"); + return { channel: "ax-platform", ok: false, error: "No target space" }; + } + + try { + const content = text ? `${text}\n\n${mediaUrl}` : mediaUrl; + const result = await callAxTool(mcpEndpoint, authToken, "messages", { + action: "send", + content, + space_id: spaceId, + }) as Record; + + logger.info(`[ax-platform] Outbound media sent to ${spaceId}: ${mediaUrl}`); + return { channel: "ax-platform", ok: true, messageId: result.message_id }; + } catch (err) { + logger.error(`[ax-platform] Outbound sendMedia error: ${err}`); + return { channel: "ax-platform", ok: false, error: String(err) }; + } }, }, @@ -627,6 +827,82 @@ export function createAxChannel(config: { dispatchStates.clear(); }, }, + + // Status adapter for dashboard health checks + status: { + defaultRuntime: { + accountId: "default", + running: false, + lastStartAt: null, + lastStopAt: null, + lastError: null, + }, + + buildAccountSnapshot: ({ account, runtime, probe, audit }: { + account: any; cfg?: any; runtime?: any; probe?: any; audit?: any; + }) => ({ + accountId: account?.accountId || "default", + name: account?.handle ? `@${account.handle.replace(/^@/, "")}` : account?.accountId, + enabled: account?.enabled !== false, + configured: Boolean(account?.agentId && account?.secret), + running: runtime?.running ?? false, + connected: probe?.ok ?? undefined, + lastStartAt: runtime?.lastStartAt ?? null, + lastStopAt: runtime?.lastStopAt ?? null, + lastError: runtime?.lastError ?? null, + lastInboundAt: runtime?.lastInboundAt ?? null, + lastOutboundAt: runtime?.lastOutboundAt ?? null, + mode: "webhook", + probe, + audit, + }), + + buildChannelSummary: ({ snapshot }: { snapshot: any; account?: any; cfg?: any; defaultAccountId?: string }) => ({ + configured: snapshot?.configured ?? false, + running: snapshot?.running ?? false, + lastError: snapshot?.lastError ?? null, + }), + + probeAccount: async ({ account, timeoutMs }: { account: any; timeoutMs: number; cfg?: any }) => { + if (!account?.agentId || !account?.secret) { + return { ok: false, error: "Missing agentId or secret" }; + } + + // Only probe prod agents against paxai.app + if (account.env && account.env !== "prod") { + return { ok: true, mode: "local", skipped: true }; + } + + const token = await resolveOutboundToken(outboundConfig); + if (!token) { + return { ok: false, error: "No access token for outbound" }; + } + + // Reuse cached probe if fresh (within 60s) — avoids N calls per status refresh + if (probeCache && Date.now() - probeCache.ts < 60000) { + return probeCache.result; + } + + const mcpEndpoint = (outboundConfig as OutboundConfig).mcpEndpoint + || process.env.AX_MCP_ENDPOINT || "https://mcp.paxai.app"; + const handle = account.handle?.replace(/^@/, "") || account.accountId; + + try { + const data = await callAxTool(mcpEndpoint, token, "agents", { + search: handle, + limit: 1, + }) as any; + const found = Array.isArray(data?.agents) && data.agents.length > 0; + const result = { ok: true, mode: "webhook", agentFound: found, handle: `@${handle}` }; + probeCache = { result, ts: Date.now() }; + return result; + } catch (err: any) { + const result = { ok: false, error: String(err) }; + probeCache = { result, ts: Date.now() }; + return result; + } + }, + }, }; } @@ -641,6 +917,7 @@ export function createDispatchHandler( config: { backendUrl?: string } ) { const backendUrl = config.backendUrl || "http://localhost:8001"; + const outboundConfig = config.outbound || {}; return async (req: IncomingMessage, res: ServerResponse): Promise => { // Only handle /ax/dispatch diff --git a/extension/clawdbot.plugin.json b/extension/clawdbot.plugin.json index 3ba0674..e64cbb0 100644 --- a/extension/clawdbot.plugin.json +++ b/extension/clawdbot.plugin.json @@ -1,8 +1,8 @@ { "id": "ax-platform", "name": "aX Platform Integration", - "description": "Connect Moltbot agents to aX Platform cloud collaboration", - "version": "0.2.0", + "description": "Connect agents to aX Platform for multi-agent cloud collaboration", + "version": "0.3.0", "channels": ["ax-platform"], "configSchema": { "type": "object", @@ -10,7 +10,7 @@ "properties": { "agents": { "type": "array", - "description": "List of aX agents this gateway handles", + "description": "List of aX agents (legacy config location)", "items": { "type": "object", "properties": { @@ -25,12 +25,46 @@ "backendUrl": { "type": "string", "description": "aX backend API URL", - "default": "http://localhost:8001" + "default": "https://api.paxai.app" + }, + "outbound": { + "type": "object", + "description": "Outbound delivery configuration", + "properties": { + "mcpEndpoint": { "type": "string", "description": "MCP server URL", "default": "https://mcp.paxai.app" }, + "tokenFile": { "type": "string", "description": "Path to OAuth token JSON file" }, + "defaultSpaceId": { "type": "string", "description": "Default space ID for delivery" } + } + } + } + }, + "channelConfigSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean", "default": true }, + "accounts": { + "type": "object", + "description": "Agent accounts (accountId -> config)", + "additionalProperties": { + "type": "object", + "properties": { + "agentId": { "type": "string", "description": "Agent UUID" }, + "secret": { "type": "string", "description": "Webhook HMAC secret" }, + "handle": { "type": "string", "description": "Agent @handle" }, + "enabled": { "type": "boolean", "default": true }, + "env": { "type": "string", "description": "Environment label" } + }, + "required": ["agentId", "secret"] + } } } }, "uiHints": { - "agents": { "label": "aX Agents", "sensitive": false }, - "backendUrl": { "label": "Backend URL" } + "agents": { "label": "aX Agents (Legacy)", "sensitive": false }, + "agents.*.secret": { "label": "HMAC Secret", "sensitive": true }, + "backendUrl": { "label": "Backend URL" }, + "outbound.tokenFile": { "label": "Token File Path" }, + "outbound.mcpEndpoint": { "label": "MCP Endpoint" } } } diff --git a/extension/index.ts b/extension/index.ts index 5729c52..bafead2 100644 --- a/extension/index.ts +++ b/extension/index.ts @@ -11,6 +11,7 @@ import type { ClawdbotPluginApi, PluginRuntime } from "clawdbot/plugin-sdk"; import { createAxChannel, createDispatchHandler, setAxPlatformRuntime, getDispatchSession } from "./channel/ax-channel.js"; +import type { OutboundConfig } from "./lib/types.js"; import { logRegisteredAgents } from "./lib/auth.js"; import { axMessagesTool } from "./tools/ax-messages.js"; import { axTasksTool } from "./tools/ax-tasks.js"; @@ -28,6 +29,7 @@ interface AxPlatformConfig { env?: string; }>; backendUrl?: string; + outbound?: OutboundConfig; } const plugin = { @@ -52,6 +54,14 @@ const plugin = { }, }, backendUrl: { type: "string" as const }, + outbound: { + type: "object" as const, + properties: { + mcpEndpoint: { type: "string" as const }, + tokenFile: { type: "string" as const }, + defaultSpaceId: { type: "string" as const }, + }, + }, }, }, @@ -61,11 +71,13 @@ const plugin = { // Store runtime for channel to access setAxPlatformRuntime(api.runtime); - const config = (api.config || {}) as AxPlatformConfig; - api.logger.info(`[ax-platform] Config received: agents=${config.agents?.length || 0}, backendUrl=${config.backendUrl || 'default'}`); + // api.config is the full clawdbot.json — extract plugin-specific config + const fullConfig = (api.config || {}) as Record; + const config = (fullConfig?.plugins?.entries?.["ax-platform"]?.config || {}) as AxPlatformConfig; + api.logger.info(`[ax-platform] Config received: agents=${config.agents?.length || 0}, backendUrl=${config.backendUrl || 'default'}, outbound=${config.outbound?.tokenFile ? 'configured' : 'none'}`); // Register the aX Platform channel - const channel = createAxChannel(config); + const channel = createAxChannel({ ...config, outbound: config.outbound }); api.registerChannel({ plugin: channel }); api.logger.info("[ax-platform] Channel registered: ax-platform"); diff --git a/extension/lib/api.ts b/extension/lib/api.ts index 5dfd72e..054ec3d 100644 --- a/extension/lib/api.ts +++ b/extension/lib/api.ts @@ -35,7 +35,10 @@ export async function sendProgressUpdate( } /** - * Call aX MCP tool via backend API + * Call aX MCP tool via JSON-RPC (MCP protocol) + * + * The MCP server speaks JSON-RPC at {mcpEndpoint}/mcp. + * We send a `tools/call` request with the tool name and arguments. */ export async function callAxTool( mcpEndpoint: string, @@ -43,18 +46,93 @@ export async function callAxTool( toolName: string, params: Record ): Promise { - const response = await fetch(`${mcpEndpoint}/tools/${toolName}`, { + if (!mcpEndpoint) throw new Error("callAxTool: mcpEndpoint is required"); + if (!authToken) throw new Error("callAxTool: authToken is required"); + if (!toolName) throw new Error("callAxTool: toolName is required"); + + // Ensure endpoint ends with /mcp (the JSON-RPC endpoint) + const rpcUrl = mcpEndpoint.endsWith("/mcp") + ? mcpEndpoint + : `${mcpEndpoint.replace(/\/+$/, "")}/mcp`; + + const body = { + jsonrpc: "2.0", + id: `ax-plugin-${Date.now()}`, + method: "tools/call", + params: { + name: toolName, + arguments: params, + }, + }; + + const response = await fetch(rpcUrl, { method: "POST", headers: { "Content-Type": "application/json", - "Authorization": `Bearer ${authToken}`, + Accept: "application/json, text/event-stream", + Authorization: `Bearer ${authToken}`, }, - body: JSON.stringify(params), + body: JSON.stringify(body), }); if (!response.ok) { - throw new Error(`aX tool ${toolName} failed: ${response.status}`); + const errText = await response.text().catch(() => ""); + throw new Error(`aX tool ${toolName} failed: ${response.status} ${errText}`); + } + + const contentType = response.headers.get("content-type") || ""; + + // Handle SSE (streaming) responses — collect until we get a JSON-RPC result + if (contentType.includes("text/event-stream")) { + const text = await response.text(); + const lines = text.split("\n"); + for (let i = lines.length - 1; i >= 0; i--) { + const line = lines[i].trim(); + if (line.startsWith("data:")) { + try { + const eventData = JSON.parse(line.slice(5).trim()); + if (eventData.result) { + return extractToolResult(eventData.result); + } + if (eventData.error) { + throw new Error(`MCP error: ${eventData.error.message || JSON.stringify(eventData.error)}`); + } + } catch (parseErr) { + if (parseErr instanceof Error && parseErr.message.startsWith("MCP error:")) throw parseErr; + // Not valid JSON — skip + } + } + } + throw new Error(`aX tool ${toolName}: no result in SSE stream`); } - return response.json(); + // Handle plain JSON-RPC response + const rpcResult = await response.json(); + if (rpcResult.error) { + throw new Error(`MCP error: ${rpcResult.error.message || JSON.stringify(rpcResult.error)}`); + } + return extractToolResult(rpcResult.result); +} + +/** + * Extract text content from an MCP tools/call result. + * + * MCP returns: { content: [{ type: "text", text: "..." }, ...] } + * This extracts text parts, tries JSON parse, falls back to raw string. + */ +function extractToolResult(result: unknown): unknown { + if (!result || typeof result !== "object") return result; + const r = result as Record; + if (Array.isArray(r.content)) { + const textParts = (r.content as Array<{ type?: string; text?: string }>) + .filter(c => c.type === "text" && c.text) + .map(c => c.text!); + const joined = textParts.join("\n"); + try { + return JSON.parse(joined); + } catch { + return joined || result; + } + } + return result; } diff --git a/extension/lib/types.ts b/extension/lib/types.ts index 505c090..3485cde 100644 --- a/extension/lib/types.ts +++ b/extension/lib/types.ts @@ -99,3 +99,10 @@ export interface DispatchSession { contextData?: ContextData; startTime: number; } + +// Outbound delivery configuration +export interface OutboundConfig { + mcpEndpoint?: string; // Default: https://mcp.paxai.app + tokenFile?: string; // Path to token JSON file + defaultSpaceId?: string; // Default space for delivery +} diff --git a/setup.sh b/setup.sh index 765ddd4..3b6142f 100755 --- a/setup.sh +++ b/setup.sh @@ -160,13 +160,28 @@ cmd_sync() { cd "$SCRIPT_DIR" log_info "Updating clawdbot.json..." - UPDATED_CONFIG=$(cat "$CONFIG_FILE" | jq --argjson agents "$AGENTS_JSON" ' + # Resolve outbound token file (first .ax-token.json found in workspaces) + local token_file="" + for ws in "$HOME/.clawdbot/workspaces"/*/.ax-token.json; do + if [[ -f "$ws" ]]; then + token_file="$ws" + break + fi + done + + UPDATED_CONFIG=$(cat "$CONFIG_FILE" | jq --argjson agents "$AGENTS_JSON" --arg tokenFile "${token_file}" ' .plugins.entries["ax-platform"].enabled = true | .plugins.entries["ax-platform"].config.agents = $agents | - .plugins.entries["ax-platform"].config.backendUrl = "https://api.paxai.app" + .plugins.entries["ax-platform"].config.backendUrl = "https://api.paxai.app" | + .plugins.entries["ax-platform"].config.outbound.mcpEndpoint = "https://mcp.paxai.app" | + if $tokenFile != "" then .plugins.entries["ax-platform"].config.outbound.tokenFile = $tokenFile else . end ') echo "$UPDATED_CONFIG" > "$CONFIG_FILE" - log_ok "Updated plugin config" + if [[ -n "$token_file" ]]; then + log_ok "Updated plugin config (outbound token: ${token_file##*/workspaces/})" + else + log_warn "Updated plugin config (no .ax-token.json found for outbound)" + fi # Clean stale env vars from plist (prevents signature verification failures) if clean_plist_env; then