diff --git a/packages/app/src/components/create-agent-modal.tsx b/packages/app/src/components/create-agent-modal.tsx index 9c150b5e9..6f62c08a8 100644 --- a/packages/app/src/components/create-agent-modal.tsx +++ b/packages/app/src/components/create-agent-modal.tsx @@ -983,16 +983,20 @@ function AgentFlowModal({ ); return; } + + // Set processing state BEFORE stopping to ensure UI updates properly + setIsDictationProcessing(true); + try { const audioData = await dictationRecorder.stop(); setIsDictating(false); setDictationVolume(0); if (!audioData) { + setIsDictationProcessing(false); return; } const requestId = generateMessageId(); dictationRequestIdRef.current = requestId; - setIsDictationProcessing(true); await sendAgentAudio( DICTATION_AGENT_ID, audioData, diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 207f8adfc..55fa21eef 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -1,55 +1,131 @@ -import { create } from "zustand"; +import { useSyncExternalStore } from "react"; import type { SessionContextValue } from "@/contexts/session-context"; // SessionData mirrors SessionContextValue so consumers can subscribe to a single source of truth. export type SessionData = SessionContextValue; -interface SessionStore { +interface SessionStoreState { sessions: Record; +} + +interface SessionStore extends SessionStoreState { setSession: (serverId: string, data: SessionData) => void; updateSession: (serverId: string, partial: Partial) => void; clearSession: (serverId: string) => void; getSession: (serverId: string) => SessionData | undefined; } -export const useSessionStore = create((set, get) => ({ - sessions: {}, - setSession: (serverId, data) => { - set((state) => ({ +type SessionListener = () => void; + +let storeState: SessionStoreState = { sessions: {} }; +const listeners = new Set(); + +const emit = () => { + for (const listener of listeners) { + listener(); + } +}; + +const updateStoreState = (updater: (prev: SessionStoreState) => SessionStoreState) => { + const next = updater(storeState); + if (next === storeState) { + return; + } + storeState = next; + emit(); +}; + +const setSession: SessionStore["setSession"] = (serverId, data) => { + updateStoreState((prev) => { + if (prev.sessions[serverId] === data) { + return prev; + } + return { sessions: { - ...state.sessions, + ...prev.sessions, [serverId]: data, }, - })); - }, - updateSession: (serverId, partial) => { - set((state) => { - const existing = state.sessions[serverId]; - const next: SessionData | undefined = existing - ? { ...existing, ...partial, serverId } - : (partial.serverId ? { ...(partial as SessionData), serverId } : undefined); + }; + }); +}; - if (!next) { - return state; - } +const updateSession: SessionStore["updateSession"] = (serverId, partial) => { + updateStoreState((prev) => { + const existing = prev.sessions[serverId]; + const next: SessionData | undefined = existing + ? { ...existing, ...partial, serverId } + : partial.serverId + ? ({ ...(partial as SessionData), serverId }) + : undefined; - return { - sessions: { - ...state.sessions, - [serverId]: next, - }, - }; - }); - }, - clearSession: (serverId) => { - set((state) => { - if (!(serverId in state.sessions)) { - return state; - } - const nextSessions = { ...state.sessions }; - delete nextSessions[serverId]; - return { sessions: nextSessions }; - }); - }, - getSession: (serverId) => get().sessions[serverId], -})); + if (!next) { + return prev; + } + + if (existing && shallowEqual(existing, next)) { + return prev; + } + + return { + sessions: { + ...prev.sessions, + [serverId]: next, + }, + }; + }); +}; + +const clearSession: SessionStore["clearSession"] = (serverId) => { + updateStoreState((prev) => { + if (!(serverId in prev.sessions)) { + return prev; + } + const nextSessions = { ...prev.sessions }; + delete nextSessions[serverId]; + return { sessions: nextSessions }; + }); +}; + +const getSession: SessionStore["getSession"] = (serverId) => { + return storeState.sessions[serverId]; +}; + +const subscribe = (listener: SessionListener) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +}; + +const buildSnapshot = (): SessionStore => ({ + sessions: storeState.sessions, + setSession, + updateSession, + clearSession, + getSession, +}); + +const shallowEqual = (left: SessionData, right: SessionData): boolean => { + if (left === right) { + return true; + } + const leftEntries = Object.entries(left); + const rightEntries = Object.entries(right); + if (leftEntries.length !== rightEntries.length) { + return false; + } + for (const [key, value] of leftEntries) { + if (right[key as keyof SessionData] !== value) { + return false; + } + } + return true; +}; + +export function useSessionStore(selector: (state: SessionStore) => T): T { + return useSyncExternalStore( + subscribe, + () => selector(buildSnapshot()), + () => selector(buildSnapshot()) + ); +} diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index f632f72fe..7b32bad6c 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -68,6 +68,15 @@ export type AgentManagerOptions = { idFactory?: () => string; }; +export type WaitForAgentOptions = { + signal?: AbortSignal; +}; + +export type WaitForAgentResult = { + status: AgentLifecycleStatus; + permission: AgentPermissionRequest | null; +}; + type ManagedAgent = { id: string; provider: AgentProvider; @@ -97,6 +106,28 @@ type SubscriptionRecord = { }; const DEFAULT_MAX_TIMELINE_ITEMS = 2000; +const BUSY_STATUSES: AgentLifecycleStatus[] = [ + "initializing", + "running", +]; + +function isAgentBusy(status: AgentLifecycleStatus): boolean { + return BUSY_STATUSES.includes(status); +} + +function createAbortError( + signal: AbortSignal | undefined, + fallbackMessage: string +): Error { + const reason = signal?.reason; + const message = + typeof reason === "string" + ? reason + : reason instanceof Error + ? reason.message + : fallbackMessage; + return Object.assign(new Error(message), { name: "AbortError" }); +} export class AgentManager { private readonly clients = new Map(); @@ -460,6 +491,105 @@ export class AgentManager { await this.primeHistory(agent); } + async waitForAgentEvent( + agentId: string, + options?: WaitForAgentOptions + ): Promise { + const snapshot = this.getAgent(agentId); + if (!snapshot) { + throw new Error(`Agent ${agentId} not found`); + } + + const immediatePermission = snapshot.pendingPermissions[0] ?? null; + if (immediatePermission) { + return { status: snapshot.status, permission: immediatePermission }; + } + + if (!isAgentBusy(snapshot.status)) { + return { status: snapshot.status, permission: null }; + } + + if (options?.signal?.aborted) { + throw createAbortError(options.signal, "wait_for_agent aborted"); + } + + return await new Promise((resolve, reject) => { + let currentStatus: AgentLifecycleStatus = snapshot.status; + const cleanupFns: Array<() => void> = []; + + const cleanup = () => { + while (cleanupFns.length) { + const fn = cleanupFns.pop(); + try { + fn?.(); + } catch { + // ignore cleanup errors + } + } + }; + + const finish = (permission: AgentPermissionRequest | null) => { + cleanup(); + resolve({ status: currentStatus, permission }); + }; + + const unsubscribe = this.subscribe( + (event) => { + if (event.type === "agent_state") { + currentStatus = event.agent.status; + const pending = event.agent.pendingPermissions[0] ?? null; + if (pending) { + finish(pending); + return; + } + if (!isAgentBusy(event.agent.status)) { + finish(null); + } + return; + } + + if (event.type !== "agent_stream") { + return; + } + + switch (event.event.type) { + case "permission_requested": { + currentStatus = "running"; + finish(event.event.request); + break; + } + case "turn_completed": { + currentStatus = "idle"; + finish(null); + break; + } + case "turn_failed": { + currentStatus = "error"; + finish(null); + break; + } + default: + break; + } + }, + { agentId, replayState: true } + ); + cleanupFns.push(unsubscribe); + + if (options?.signal) { + const abortHandler = () => { + cleanup(); + reject(createAbortError(options.signal, "wait_for_agent aborted")); + }; + + options.signal.addEventListener("abort", abortHandler, { once: true }); + cleanupFns.push(() => + options.signal?.removeEventListener("abort", abortHandler) + ); + } + }); + } + private async registerSession( session: AgentSession, config: AgentSessionConfig, diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index 3fef065ba..fc77e696b 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -6,12 +6,11 @@ import { ensureValidJson } from "../json-utils.js"; import type { AgentPromptInput, AgentProvider, - AgentPermissionRequest, } from "./agent-sdk-types.js"; import type { - AgentLifecycleStatus, AgentManager, AgentSnapshot, + WaitForAgentResult, } from "./agent-manager.js"; import { AgentPermissionRequestPayloadSchema, @@ -73,10 +72,6 @@ function startAgentRun( })(); } -function isAgentBusy(status: AgentLifecycleStatus): boolean { - return status === "running" || status === "initializing"; -} - async function resolveAgentTitle( agentRegistry: AgentRegistry, agentId: string @@ -101,106 +96,6 @@ async function serializeSnapshotWithMetadata( return serializeAgentSnapshot(snapshot, { title }); } -async function waitForAgentEvent( - agentManager: AgentManager, - agentId: string, - tracker: WaitForAgentTracker, - signal?: AbortSignal -): Promise<{ - status: AgentLifecycleStatus; - permission: AgentPermissionRequest | null; -}> { - const snapshot = agentManager.getAgent(agentId); - if (!snapshot) { - throw new Error(`Agent ${agentId} not found`); - } - - const existingPermission = snapshot.pendingPermissions[0] ?? null; - if (existingPermission) { - return { status: snapshot.status, permission: existingPermission }; - } - - if (!isAgentBusy(snapshot.status)) { - return { status: snapshot.status, permission: null }; - } - - return await new Promise((resolve, reject) => { - let currentStatus: AgentLifecycleStatus = snapshot.status; - const cleanupFns: Array<() => void> = []; - - const cleanup = () => { - while (cleanupFns.length) { - const fn = cleanupFns.pop(); - try { - fn?.(); - } catch { - // ignore cleanup errors - } - } - }; - - const finish = (permission: AgentPermissionRequest | null) => { - cleanup(); - resolve({ status: currentStatus, permission }); - }; - - const unregister = tracker.register(agentId, (reason) => { - cleanup(); - reject( - Object.assign( - new Error(reason ?? "wait_for_agent cancelled"), - { - name: "AbortError", - } - ) - ); - }); - cleanupFns.push(unregister); - - const unsubscribe = agentManager.subscribe( - (event) => { - if (event.type === "agent_state") { - currentStatus = event.agent.status; - if (!isAgentBusy(event.agent.status)) { - const pending = event.agent.pendingPermissions[0] ?? null; - finish(pending); - } - return; - } - - if ( - event.type === "agent_stream" && - event.event.type === "permission_requested" - ) { - currentStatus = "running"; - finish(event.event.request); - } - }, - { agentId, replayState: false } - ); - cleanupFns.push(unsubscribe); - - if (signal) { - const abort = () => { - cleanup(); - reject( - Object.assign(new Error("wait_for_agent aborted"), { - name: "AbortError", - }) - ); - }; - - if (signal.aborted) { - abort(); - return; - } - - signal.addEventListener("abort", abort, { once: true }); - cleanupFns.push(() => signal.removeEventListener("abort", abort)); - } - }); -} - function buildActivityPayload( agentManager: AgentManager, agentId: string @@ -346,23 +241,64 @@ export async function createAgentMcpServer( }, }, async ({ agentId }, { signal }) => { - const result = await waitForAgentEvent( - agentManager, - agentId, - waitTracker, - signal - ); - const activity = buildActivityPayload(agentManager, agentId); + const abortController = new AbortController(); + const cleanupFns: Array<() => void> = []; - return { - content: [], - structuredContent: ensureValidJson({ - agentId, - status: result.status, - permission: result.permission, - activity, - }), + const cleanup = () => { + while (cleanupFns.length) { + const fn = cleanupFns.pop(); + try { + fn?.(); + } catch { + // ignore cleanup errors + } + } }; + + const forwardExternalAbort = () => { + if (!abortController.signal.aborted) { + const reason = signal?.reason ?? new Error("wait_for_agent aborted"); + abortController.abort(reason); + } + }; + + if (signal) { + if (signal.aborted) { + forwardExternalAbort(); + } else { + signal.addEventListener("abort", forwardExternalAbort, { once: true }); + cleanupFns.push(() => + signal.removeEventListener("abort", forwardExternalAbort) + ); + } + } + + const unregister = waitTracker.register(agentId, (reason) => { + if (!abortController.signal.aborted) { + abortController.abort(new Error(reason ?? "wait_for_agent cancelled")); + } + }); + cleanupFns.push(unregister); + + try { + const result: WaitForAgentResult = + await agentManager.waitForAgentEvent(agentId, { + signal: abortController.signal, + }); + const activity = buildActivityPayload(agentManager, agentId); + + return { + content: [], + structuredContent: ensureValidJson({ + agentId, + status: result.status, + permission: result.permission, + activity, + }), + }; + } finally { + cleanup(); + } } ); diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index b5961b8a7..d06d6504f 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -75,6 +75,29 @@ type ClaudeAgentConfig = AgentSessionConfig & { provider: "claude" }; export type ClaudeContentChunk = { type: string; [key: string]: any }; +type AgentControlMcpConfig = { + url: string; + headers?: Record; +}; + +type ClaudeAgentClientOptions = { + defaults?: { agents?: Record }; + agentControlMcp?: AgentControlMcpConfig; +}; + +type ClaudeAgentSessionOptions = { + defaults?: { agents?: Record }; + handle?: AgentPersistenceHandle; + agentControlMcp?: AgentControlMcpConfig; +}; + +const DEFAULT_AGENT_CONTROL_MCP: AgentControlMcpConfig = { + url: "http://127.0.0.1:6767/mcp/agents", + headers: { + Authorization: "Basic bW86Ym8=", + }, +}; + export function extractUserMessageText( content: string | ClaudeContentChunk[] ): string | null { @@ -142,11 +165,20 @@ export class ClaudeAgentClient implements AgentClient { readonly provider = "claude" as const; readonly capabilities = CLAUDE_CAPABILITIES; - constructor(private readonly defaults?: { agents?: Record }) {} + private readonly defaults?: { agents?: Record }; + private readonly agentControlMcp?: AgentControlMcpConfig; + + constructor(options?: ClaudeAgentClientOptions) { + this.defaults = options?.defaults; + this.agentControlMcp = options?.agentControlMcp; + } async createSession(config: AgentSessionConfig): Promise { const claudeConfig = this.assertConfig(config); - return new ClaudeAgentSession(claudeConfig, this.defaults); + return new ClaudeAgentSession(claudeConfig, { + defaults: this.defaults, + agentControlMcp: this.agentControlMcp, + }); } async resumeSession( @@ -160,7 +192,11 @@ export class ClaudeAgentClient implements AgentClient { } const mergedConfig = { ...merged, provider: "claude" } as AgentSessionConfig; const claudeConfig = this.assertConfig(mergedConfig); - return new ClaudeAgentSession(claudeConfig, this.defaults, handle); + return new ClaudeAgentSession(claudeConfig, { + defaults: this.defaults, + agentControlMcp: this.agentControlMcp, + handle, + }); } async listPersistedAgents(options?: ListPersistedAgentsOptions): Promise { @@ -199,6 +235,7 @@ class ClaudeAgentSession implements AgentSession { private readonly config: ClaudeAgentConfig; private readonly defaults?: { agents?: Record }; + private readonly agentControlMcp?: AgentControlMcpConfig; private query: Query | null = null; private input: Pushable | null = null; private claudeSessionId: string | null; @@ -220,11 +257,12 @@ class ClaudeAgentSession implements AgentSession { constructor( config: ClaudeAgentConfig, - defaults?: { agents?: Record }, - handle?: AgentPersistenceHandle + options?: ClaudeAgentSessionOptions ) { this.config = config; - this.defaults = defaults; + this.defaults = options?.defaults; + this.agentControlMcp = options?.agentControlMcp; + const handle = options?.handle; this.claudeSessionId = handle?.sessionId ?? handle?.nativeHandle ?? null; this.pendingLocalId = this.claudeSessionId ?? `claude-${randomUUID()}`; this.persistence = handle ?? null; @@ -459,9 +497,24 @@ class ClaudeAgentSession implements AgentSession { ...this.config.extra?.claude, }; + // Always include the agent-control MCP server so agents can launch other agents + const agentControlConfig = this.agentControlMcp ?? DEFAULT_AGENT_CONTROL_MCP; + const agentControlMcp: Record = { + "agent-control": { + type: "http", + url: agentControlConfig.url, + ...(agentControlConfig.headers ? { headers: agentControlConfig.headers } : {}), + }, + }; + if (this.config.mcpServers) { - base.mcpServers = this.normalizeMcpServers(this.config.mcpServers); + const normalizedUserServers = this.normalizeMcpServers(this.config.mcpServers); + // Merge user-provided MCP servers with agent-control, user servers take precedence + base.mcpServers = { ...agentControlMcp, ...normalizedUserServers }; + } else { + base.mcpServers = agentControlMcp; } + if (this.config.model) { base.model = this.config.model; } diff --git a/packages/server/src/server/index.ts b/packages/server/src/server/index.ts index 3537ddb3e..40b84f7a3 100644 --- a/packages/server/src/server/index.ts +++ b/packages/server/src/server/index.ts @@ -2,6 +2,7 @@ import "dotenv/config"; import express from "express"; import basicAuth from "express-basic-auth"; import { createServer as createHTTPServer } from "http"; +import { randomUUID } from "node:crypto"; import { VoiceAssistantWebSocketServer } from "./websocket-server.js"; import { initializeSTT, type STTConfig } from "./agent/stt-openai.js"; import { initializeTTS } from "./agent/tts-openai.js"; @@ -15,6 +16,16 @@ import { attachAgentRegistryPersistence, restorePersistedAgents, } from "./persistence-hooks.js"; +import { createAgentMcpServer } from "./agent/mcp-server.js"; +import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; +import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; + +type AgentMcpTransportMap = Map< + string, + StreamableHTTPServerTransport +>; + +const BASIC_AUTH_USERS = { mo: "bo" } as const; function createServer() { const app = express(); @@ -25,7 +36,7 @@ function createServer() { // Basic authentication (skip for /public routes) app.use( basicAuth({ - users: { mo: "bo" }, + users: BASIC_AUTH_USERS, challenge: true, realm: "Voice Assistant", }) @@ -66,6 +77,16 @@ function createServer() { async function main() { const port = parseInt(process.env.PORT || "6767", 10); + const agentMcpRoute = "/mcp/agents"; + const agentMcpUrl = `http://127.0.0.1:${port}${agentMcpRoute}`; + const [agentMcpUser, agentMcpPassword] = + Object.entries(BASIC_AUTH_USERS)[0] ?? []; + const agentMcpAuthHeader = + agentMcpUser && agentMcpPassword + ? `Basic ${Buffer.from( + `${agentMcpUser}:${agentMcpPassword}` + ).toString("base64")}` + : undefined; const app = createServer(); const httpServer = createHTTPServer(app); @@ -74,7 +95,14 @@ async function main() { const agentRegistry = new AgentRegistry(); const agentManager = new AgentManager({ clients: { - claude: new ClaudeAgentClient(), + claude: new ClaudeAgentClient({ + agentControlMcp: { + url: agentMcpUrl, + ...(agentMcpAuthHeader + ? { headers: { Authorization: agentMcpAuthHeader } } + : {}), + }, + }), codex: new CodexAgentClient(), }, }); @@ -84,11 +112,112 @@ async function main() { await restorePersistedAgents(agentManager, agentRegistry); console.log("✓ Global agent manager initialized with persisted agents"); + const agentMcpServer = await createAgentMcpServer({ + agentManager, + agentRegistry, + }); + const agentMcpTransports: AgentMcpTransportMap = new Map(); + + const createAgentMcpTransport = async () => { + const transport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: (sessionId) => { + agentMcpTransports.set(sessionId, transport); + console.log(`[Agent MCP] Session initialized: ${sessionId}`); + }, + onsessionclosed: (sessionId) => { + agentMcpTransports.delete(sessionId); + console.log(`[Agent MCP] Session closed: ${sessionId}`); + }, + enableDnsRebindingProtection: true, + allowedHosts: [`127.0.0.1:${port}`, `localhost:${port}`], + }); + transport.onclose = () => { + if (transport.sessionId) { + agentMcpTransports.delete(transport.sessionId); + } + }; + transport.onerror = (error) => { + console.error("[Agent MCP] Transport error:", error); + }; + await agentMcpServer.connect(transport); + return transport; + }; + + const handleAgentMcpRequest: express.RequestHandler = async ( + req, + res + ) => { + try { + const sessionId = req.header("mcp-session-id"); + let transport = sessionId + ? agentMcpTransports.get(sessionId) + : undefined; + + if (!transport) { + if (req.method !== "POST") { + res.status(400).json({ + jsonrpc: "2.0", + error: { + code: -32000, + message: "Missing or invalid MCP session", + }, + id: null, + }); + return; + } + if (!isInitializeRequest(req.body)) { + res.status(400).json({ + jsonrpc: "2.0", + error: { + code: -32000, + message: "Initialization request expected", + }, + id: null, + }); + return; + } + transport = await createAgentMcpTransport(); + } + + await transport.handleRequest( + req as any, + res as any, + req.body + ); + } catch (error) { + console.error("[Agent MCP] Failed to handle request:", error); + if (!res.headersSent) { + res.status(500).json({ + jsonrpc: "2.0", + error: { + code: -32603, + message: "Internal MCP server error", + }, + id: null, + }); + } + } + }; + + app.post(agentMcpRoute, handleAgentMcpRequest); + app.get(agentMcpRoute, handleAgentMcpRequest); + app.delete(agentMcpRoute, handleAgentMcpRequest); + console.log(`✓ Agent MCP server mounted at ${agentMcpRoute}`); + // Initialize WebSocket server const wsServer = new VoiceAssistantWebSocketServer( httpServer, agentManager, - agentRegistry + agentRegistry, + { + agentMcpUrl, + agentMcpHeaders: agentMcpAuthHeader + ? { + Authorization: agentMcpAuthHeader, + } + : undefined, + } ); // Initialize OpenAI client diff --git a/packages/server/src/server/messages.ts b/packages/server/src/server/messages.ts index 2f478c390..82128cea6 100644 --- a/packages/server/src/server/messages.ts +++ b/packages/server/src/server/messages.ts @@ -244,6 +244,59 @@ export const AgentSnapshotPayloadSchema = z.object({ title: z.string().nullable(), }); +function sanitizeOptionalJson(value: unknown): unknown { + if (value === undefined) { + return undefined; + } + if (value === null) { + return null; + } + if (Array.isArray(value)) { + return value + .map((item) => sanitizeOptionalJson(item)) + .filter((item) => item !== undefined); + } + if (value instanceof Date) { + return value.toISOString(); + } + if (typeof value === "object") { + const result: Record = {}; + for (const [key, val] of Object.entries(value as Record)) { + const sanitized = sanitizeOptionalJson(val); + if (sanitized !== undefined) { + result[key] = sanitized; + } + } + return Object.keys(result).length ? result : undefined; + } + return value; +} + +function sanitizeOptionalJsonValue(value: T | undefined): T | undefined { + const sanitized = sanitizeOptionalJson(value); + return sanitized === undefined ? undefined : (sanitized as T); +} + +function sanitizePersistenceHandle( + handle: AgentPersistenceHandle | null +): AgentPersistenceHandle | null { + if (!handle) { + return null; + } + const sanitized: AgentPersistenceHandle = { + provider: handle.provider, + sessionId: handle.sessionId, + }; + if (handle.nativeHandle !== undefined) { + sanitized.nativeHandle = handle.nativeHandle; + } + const metadata = sanitizeOptionalJsonValue(handle.metadata); + if (metadata !== undefined) { + sanitized.metadata = metadata; + } + return sanitized; +} + export type AgentStreamEventPayload = z.infer< typeof AgentStreamEventPayloadSchema >; @@ -252,15 +305,35 @@ export function serializeAgentSnapshot( snapshot: AgentSnapshot, options?: { title?: string | null } ): AgentSnapshotPayload { - const { createdAt, updatedAt, lastUserMessageAt, ...rest } = snapshot; - return { - ...rest, + const payload: AgentSnapshotPayload = { + id: snapshot.id, + provider: snapshot.provider, + cwd: snapshot.cwd, model: snapshot.model, - createdAt: createdAt.toISOString(), - updatedAt: updatedAt.toISOString(), - lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null, + createdAt: snapshot.createdAt.toISOString(), + updatedAt: snapshot.updatedAt.toISOString(), + lastUserMessageAt: snapshot.lastUserMessageAt + ? snapshot.lastUserMessageAt.toISOString() + : null, + status: snapshot.status, + sessionId: snapshot.sessionId, + capabilities: snapshot.capabilities, + currentModeId: snapshot.currentModeId, + availableModes: snapshot.availableModes, + pendingPermissions: snapshot.pendingPermissions, + persistence: sanitizePersistenceHandle(snapshot.persistence), title: options?.title ?? null, }; + + const lastUsage = sanitizeOptionalJsonValue(snapshot.lastUsage); + if (lastUsage !== undefined) { + payload.lastUsage = lastUsage; + } + if (snapshot.lastError !== undefined) { + payload.lastError = snapshot.lastError; + } + + return payload; } export function serializeAgentStreamEvent( diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 4b94ac58a..fe72b9206 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -34,9 +34,9 @@ import { buildSessionConfig, } from "./persistence-hooks.js"; import { experimental_createMCPClient } from "ai"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import { createTerminalMcpServer } from "./terminal-mcp/index.js"; -import { createAgentMcpServer } from "./agent/mcp-server.js"; import { fetchProviderModelCatalog } from "./agent/model-catalog.js"; import { AgentManager } from "./agent/agent-manager.js"; import type { AgentSnapshot } from "./agent/agent-manager.js"; @@ -66,6 +66,11 @@ import { validateBranchSlug, } from "../utils/worktree.js"; +type AgentMcpClientConfig = { + agentMcpUrl: string; + agentMcpHeaders?: Record; +}; + const execAsync = promisify(exec); const ACTIVE_TITLE_GENERATIONS = new Set(); let restartRequested = false; @@ -223,6 +228,7 @@ export class Session { private agentTools: ToolSet | null = null; private agentManager: AgentManager; private readonly agentRegistry: AgentRegistry; + private readonly agentMcpConfig: AgentMcpClientConfig; private agentTitleCache: Map = new Map(); private unsubscribeAgentEvents: (() => void) | null = null; @@ -231,6 +237,7 @@ export class Session { onMessage: (msg: SessionOutboundMessage) => void, agentManager: AgentManager, agentRegistry: AgentRegistry, + agentMcpConfig: AgentMcpClientConfig, options?: { conversationId?: string; initialMessages?: ModelMessage[]; @@ -241,6 +248,7 @@ export class Session { this.onMessage = onMessage; this.agentManager = agentManager; this.agentRegistry = agentRegistry; + this.agentMcpConfig = agentMcpConfig; this.abortController = new AbortController(); // Initialize conversation history @@ -434,18 +442,19 @@ export class Session { */ private async initializeAgentMcp(): Promise { try { - const server = await createAgentMcpServer({ - agentManager: this.agentManager, - agentRegistry: this.agentRegistry, - }); - - const [clientTransport, serverTransport] = - InMemoryTransport.createLinkedPair(); - - await server.connect(serverTransport); + const transport = new StreamableHTTPClientTransport( + new URL(this.agentMcpConfig.agentMcpUrl), + this.agentMcpConfig.agentMcpHeaders + ? { + requestInit: { + headers: this.agentMcpConfig.agentMcpHeaders, + }, + } + : undefined + ); this.agentMcpClient = await experimental_createMCPClient({ - transport: clientTransport, + transport, }); this.agentTools = (await this.agentMcpClient.tools()) as ToolSet; diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index e3b782de1..7a8a43338 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -13,6 +13,11 @@ import { loadConversation } from "./persistence.js"; import { AgentManager } from "./agent/agent-manager.js"; import { AgentRegistry } from "./agent/agent-registry.js"; +type AgentMcpClientConfig = { + agentMcpUrl: string; + agentMcpHeaders?: Record; +}; + /** * WebSocket server that routes messages between clients and their sessions. * This is a thin transport layer with no business logic. @@ -24,14 +29,17 @@ export class VoiceAssistantWebSocketServer { private clientIdCounter: number = 0; private agentManager: AgentManager; private agentRegistry: AgentRegistry; + private readonly agentMcpConfig: AgentMcpClientConfig; constructor( server: HTTPServer, agentManager: AgentManager, - agentRegistry: AgentRegistry + agentRegistry: AgentRegistry, + agentMcpConfig: AgentMcpClientConfig ) { this.agentManager = agentManager; this.agentRegistry = agentRegistry; + this.agentMcpConfig = agentMcpConfig; this.wss = new WebSocketServer({ server, path: "/ws" }); this.wss.on("connection", (ws, request) => { @@ -79,6 +87,7 @@ export class VoiceAssistantWebSocketServer { }, this.agentManager, this.agentRegistry, + this.agentMcpConfig, { conversationId, initialMessages: initialMessages || undefined, diff --git a/test-mcp-call.mjs b/test-mcp-call.mjs new file mode 100755 index 000000000..c34f977c0 --- /dev/null +++ b/test-mcp-call.mjs @@ -0,0 +1,87 @@ +#!/usr/bin/env node + +const CLAUDE_AGENT_ID = 'f7644b18-7fb9-4491-80f3-788d0d27a119'; +const CODEX_AGENT_ID = 'd7c360ea-1579-4091-b224-ee8442400823'; + +const url = 'http://127.0.0.1:6767/mcp/agents'; +const auth = Buffer.from('mo:bo').toString('base64'); + +async function makeRequest(method, params) { + console.log(`\n[TEST] Calling ${method} with params:`, JSON.stringify(params, null, 2)); + const startTime = Date.now(); + + try { + const response = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Basic ${auth}` + }, + body: JSON.stringify({ + jsonrpc: '2.0', + id: Date.now(), + method: 'tools/call', + params: { + name: method, + arguments: params + } + }) + }); + + const elapsed = Date.now() - startTime; + console.log(`[TEST] Response received in ${elapsed}ms`); + + const data = await response.json(); + console.log(`[TEST] Response:`, JSON.stringify(data, null, 2)); + return data; + } catch (error) { + const elapsed = Date.now() - startTime; + console.error(`[TEST] Request failed after ${elapsed}ms:`, error.message); + throw error; + } +} + +async function initialize() { + console.log('[TEST] Initializing MCP session...'); + const response = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Basic ${auth}` + }, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { + protocolVersion: '2024-11-05', + capabilities: {}, + clientInfo: { + name: 'test-client', + version: '1.0.0' + } + } + }) + }); + const data = await response.json(); + console.log('[TEST] Initialized:', JSON.stringify(data, null, 2)); +} + +async function main() { + console.log('=== Testing MCP Agent Control Tools ===\n'); + + // Initialize session + await initialize(); + + // Test 1: Get status of Codex agent (should work) + console.log('\n--- Test 1: get_agent_status on Codex agent ---'); + await makeRequest('get_agent_status', { agentId: CODEX_AGENT_ID }); + + // Test 2: Get status of Claude agent (might hang) + console.log('\n--- Test 2: get_agent_status on Claude agent ---'); + await makeRequest('get_agent_status', { agentId: CLAUDE_AGENT_ID }); + + console.log('\n=== All tests complete ==='); +} + +main().catch(console.error);