Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 138 additions & 8 deletions extension/channel/ax-channel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 ───────────────────────────────────────────────────
Expand Down Expand Up @@ -571,11 +571,46 @@ export function getDispatchSession(sessionKey: string): DispatchSession | undefi
/**
* Create the aX Platform channel plugin
*/

/**
* Resolve auth token for outbound MCP calls.
* Checks token file first, then env var.
*/
async function resolveOutboundAuthToken(
outboundCfg: OutboundConfig,
logger: { error: (msg: string) => void }
): Promise<string> {
let authToken: string | undefined;

if (outboundCfg.tokenFile) {
try {
const fs = await import("node:fs");
const tokenPath = outboundCfg.tokenFile.replace(/^~/, process.env.HOME || "");
const tokenData = JSON.parse(fs.readFileSync(tokenPath, "utf-8"));
authToken = tokenData.access_token;
} catch (err) {
logger.error(`[ax-platform] Failed to read token file: ${err}`);
}
}

if (!authToken) {
authToken = process.env.AX_ACCESS_TOKEN;
}

if (!authToken) {
throw new Error("[ax-platform] Outbound failed: No access token configured. Set outbound.tokenFile or AX_ACCESS_TOKEN.");
}

return authToken;
}

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);
Expand All @@ -601,13 +636,108 @@ export function createAxChannel(config: {
},

outbound: {
deliveryMode: "direct",
deliveryMode: "direct" as const,

/**
* Resolve outbound target for aX Platform
* Accepts "space:<id>" or raw space ID
*/
resolveTarget({ to }: { to?: string; accountId?: string | null }) {
const spaceId = to?.replace(/^space:/, "") || outboundConfig?.defaultSpaceId;
if (!spaceId) {
return { ok: false as const, error: new Error("No target space. Use 'space:<id>' or configure defaultSpaceId.") };
}
return { ok: true as const, to: spaceId };
},

// 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 };
/**
* Send text to aX Platform via MCP JSON-RPC protocol.
*
* Matches Clawdbot ChannelOutboundContext signature:
* (ctx: { cfg, to, text, accountId?, ... }) => Promise<OutboundDeliveryResult>
*
* Uses callAxTool which sends proper MCP tools/call JSON-RPC to the server.
*/
async sendText({ text, to }: {
cfg: unknown;
to: string;
text: string;
mediaUrl?: string;
accountId?: string | null;
replyToId?: string | null;
threadId?: string | number | null;
deps?: unknown;
gifPlayback?: boolean;
}) {
const logger = runtime?.logger || { info: console.log, error: console.error };
const outboundCfg = outboundConfig || {};
const mcpEndpoint = outboundCfg.mcpEndpoint || process.env.AX_MCP_ENDPOINT || "https://mcp.paxai.app";
const authToken = await resolveOutboundAuthToken(outboundCfg, logger);

// Send via MCP JSON-RPC (callAxTool now uses proper protocol)
try {
const result = await callAxTool(mcpEndpoint, authToken, "messages", {
action: "send",
content: text,
space_id: to,
});

const messageId = (result && typeof result === "object" && "message_id" in result)
? String((result as Record<string, unknown>).message_id)
: `ax-${Date.now()}`;

const preview = text.length > 50 ? text.substring(0, 50) + "..." : text;
logger.info(`[ax-platform] Outbound sent to ${to}: ${preview}`);

return {
channel: "ax-platform" as const,
messageId,
chatId: to,
};
} catch (err) {
logger.error(`[ax-platform] Outbound error: ${err}`);
throw err;
}
},

/**
* Send media to aX Platform (sends caption + media link)
*/
async sendMedia({ text, to, mediaUrl }: {
cfg: unknown;
to: string;
text: string;
mediaUrl?: string;
accountId?: string | null;
replyToId?: string | null;
threadId?: string | number | null;
deps?: unknown;
gifPlayback?: boolean;
}) {
const logger = runtime?.logger || { info: console.log, error: console.error };
const outboundCfg = outboundConfig || {};
const mcpEndpoint = outboundCfg.mcpEndpoint || process.env.AX_MCP_ENDPOINT || "https://mcp.paxai.app";
const authToken = await resolveOutboundAuthToken(outboundCfg, logger);

const content = mediaUrl ? `${text}\n${mediaUrl}` : text;
try {
const result = await callAxTool(mcpEndpoint, authToken, "messages", {
action: "send",
content,
space_id: to,
});
const messageId = (result && typeof result === "object" && "message_id" in result)
? String((result as Record<string, unknown>).message_id)
: `ax-${Date.now()}`;
return {
channel: "ax-platform" as const,
messageId,
chatId: to,
};
} catch (err) {
logger.error(`[ax-platform] Outbound media error: ${err}`);
throw err;
}
},
},

Expand Down
8 changes: 6 additions & 2 deletions extension/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,15 @@

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";
import { axContextTool } from "./tools/ax-context.js";
import { axAgentsTool } from "./tools/ax-agents.js";
import { axThreadTool } from "./tools/ax-thread.js";
import { createAxProgressTool } from "./tools/ax-progress.js";
import { axDashboardTool } from "./tools/ax-dashboard.js";
import { buildMissionBriefing } from "./lib/context.js";

interface AxPlatformConfig {
Expand All @@ -28,6 +30,7 @@ interface AxPlatformConfig {
env?: string;
}>;
backendUrl?: string;
outbound?: OutboundConfig;
}

const plugin = {
Expand Down Expand Up @@ -65,7 +68,7 @@ const plugin = {
api.logger.info(`[ax-platform] Config received: agents=${config.agents?.length || 0}, backendUrl=${config.backendUrl || 'default'}`);

// 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");

Expand All @@ -79,7 +82,8 @@ const plugin = {
api.registerTool(axAgentsTool, { optional: true });
api.registerTool(axThreadTool, { optional: true });
api.registerTool(createAxProgressTool(api.runtime), { optional: true });
api.logger.info("[ax-platform] Tools registered: ax_messages, ax_tasks, ax_context, ax_agents, ax_thread, ax_progress");
api.registerTool(axDashboardTool, { optional: true });
api.logger.info("[ax-platform] Tools registered: ax_messages, ax_tasks, ax_context, ax_agents, ax_thread, ax_progress, ax_dashboard");

// Register before_agent_start hook for context injection
// Uses api.on() event pattern (like memory-lancedb plugin)
Expand Down
104 changes: 98 additions & 6 deletions extension/lib/api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,26 +35,118 @@ 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.
*
* Previous implementation used REST: `${mcpEndpoint}/tools/${toolName}`
* which doesn't exist on the MCP server. Fixed to use proper MCP JSON-RPC.
*/
export async function callAxTool(
mcpEndpoint: string,
authToken: string,
toolName: string,
params: Record<string, unknown>
): Promise<unknown> {
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();
// Parse SSE events — find the last `data:` line with a JSON-RPC result
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 — log for debugging and skip
if (process.env.DEBUG || process.env.LOG_LEVEL === "debug") {
console.debug(`[ax-platform] SSE parse skip: ${line.substring(0, 100)}`);
}
}
}
}
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 the text content from an MCP tools/call result.
*
* MCP returns results in the format:
* { content: [{ type: "text", text: "..." }, ...] }
*
* This function:
* 1. Extracts all text content parts from the content array
* 2. Joins them with newlines
* 3. Attempts to parse as JSON (for structured results like message lists)
* 4. Falls back to raw text string if not valid JSON
* 5. Returns the original result if no content array is present
*/
function extractToolResult(result: unknown): unknown {
if (!result || typeof result !== "object") return result;
const r = result as Record<string, unknown>;
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;
}
7 changes: 7 additions & 0 deletions extension/lib/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
4 changes: 2 additions & 2 deletions extension/tools/ax-agents.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ export const axAgentsTool = {
limit: Type.Optional(Type.Number({ description: "Max results", default: 25 })),
}),

async execute(_toolCallId: string, params: Record<string, unknown>, context: { sessionKey?: string }) {
const sessionKey = context.sessionKey;
async execute(_toolCallId: string, params: Record<string, unknown>, context: Record<string, unknown>) {
const sessionKey = (context.sessionKey || context.SessionKey) as string | undefined;
const session = sessionKey ? getDispatchSession(sessionKey) : undefined;

if (!session?.authToken || !session?.mcpEndpoint) {
Expand Down
4 changes: 2 additions & 2 deletions extension/tools/ax-context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@ export const axContextTool = {
prefix: Type.Optional(Type.String({ description: "Prefix filter (for list)" })),
}),

async execute(_toolCallId: string, params: Record<string, unknown>, context: { sessionKey?: string }) {
const sessionKey = context.sessionKey;
async execute(_toolCallId: string, params: Record<string, unknown>, context: Record<string, unknown>) {
const sessionKey = (context.sessionKey || context.SessionKey) as string | undefined;
const session = sessionKey ? getDispatchSession(sessionKey) : undefined;

if (!session?.authToken || !session?.mcpEndpoint) {
Expand Down
Loading