MCP Puppet Pipeline

// MCP Tools for AI-to-AI Coordination and Communication import { Pool } from 'pg'; import { InitializeConversationSchema, SendMessageSchema, PlanningSessionSchema, TaskAssignmentSchema, GetConversationStatusSchema, TaskUpdateSchema } from '../validation.js'; // Singleton database pool let dbPool: Pool | null = null; function getDbPool(): Pool { if (!dbPool) { if (!process.env.DATABASE_URL) { throw new Error('DATABASE_URL environment variable is required for AI coordination'); } dbPool = new Pool({ connectionString: process.env.DATABASE_URL, max: 10, idleTimeoutMillis: 30000, connectionTimeoutMillis: 2000, }); } return dbPool; } // Helper function to execute SQL queries async function executeQuery(query: string, params: any[] = []): Promise<any> { const pool = getDbPool(); try { const result = await pool.query(query, params); return result; } catch (error) { console.error('Database query error:', error); throw error; } } // Tool handler type type ToolHandler = (args: any) => Promise<any>; // Tool definition type interface ToolDefinition { name: string; description: string; inputSchema: any; } const aiCoordinationTools = { // Tool 1: Initialize AI conversation session initialize_ai_conversation: { name: "initialize_ai_conversation", description: "Start a structured conversation session between Claude, OpenAI, and Replit for collaborative planning and coordination.", inputSchema: { type: "object", properties: { title: { type: "string", description: "Title of the conversation session" }, objective: { type: "string", description: "Primary objective or goal of the conversation" }, orchestrator: { type: "string", description: "Which AI should orchestrate the conversation", enum: ["claude", "openai", "replit"], default: "claude" }, participants: { type: "array", items: { type: "string" }, description: "List of AI participants", default: ["claude", "openai", "replit"] }, context: { type: "object", description: "Initial context and background information", default: {} } }, required: ["title", "objective"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = InitializeConversationSchema.parse(args); // Generate unique session ID const sessionId = `ai_session_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; const query = ` INSERT INTO ai_conversations (session_id, title, objective, orchestrator, participants, context) VALUES ($1, $2, $3, $4, $5, $6) RETURNING * `; const result = await executeQuery(query, [ sessionId, validatedArgs.title, validatedArgs.objective, validatedArgs.orchestrator, JSON.stringify(validatedArgs.participants), JSON.stringify(validatedArgs.context) ]); const conversation = result.rows[0]; // Send initialization message const initMessage = `🤖 AI Conversation Initialized **Session:** ${conversation.title} **Objective:** ${conversation.objective} **Orchestrator:** ${conversation.orchestrator} **Participants:** ${validatedArgs.participants.join(', ')} The structured AI coordination session is now active. The orchestrator can begin directing the collaborative discussion and planning process.`; await executeQuery( `INSERT INTO ai_conversation_messages (conversation_id, participant, message_type, content) VALUES ($1, $2, $3, $4)`, [conversation.id, 'system', 'discussion', initMessage] ); return { success: true, session_id: sessionId, conversation_id: conversation.id, title: conversation.title, objective: conversation.objective, orchestrator: conversation.orchestrator, participants: validatedArgs.participants, message: `✅ AI conversation session "${validatedArgs.title}" initialized with session ID: ${sessionId}`, next_steps: [ `The ${conversation.orchestrator} should begin the coordinated discussion`, "Use ai_send_message to contribute to the conversation", "Use ai_planning_session to structure collaborative planning" ] }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to initialize AI conversation: ${error.message}` }; } } }, // Tool 2: Send message in AI conversation ai_send_message: { name: "ai_send_message", description: "Send a message from one AI participant to the conversation session for coordination and discussion.", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "Session ID of the conversation" }, participant: { type: "string", description: "Which AI is sending the message", enum: ["claude", "openai", "replit"] }, message_type: { type: "string", description: "Type of message being sent", enum: ["discussion", "proposal", "agreement", "objection", "question", "decision", "task_assignment"], default: "discussion" }, content: { type: "string", description: "The message content" }, context: { type: "object", description: "Additional context or structured data", default: {} }, references_message_id: { type: "integer", description: "ID of message this is responding to" } }, required: ["session_id", "participant", "content"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = SendMessageSchema.parse(args); // Get conversation ID const convQuery = `SELECT id, title FROM ai_conversations WHERE session_id = $1`; const convResult = await executeQuery(convQuery, [validatedArgs.session_id]); if (convResult.rows.length === 0) { return { success: false, error: "Conversation session not found", message: `❌ Session ${validatedArgs.session_id} not found` }; } const conversation = convResult.rows[0]; // Insert message const messageQuery = ` INSERT INTO ai_conversation_messages (conversation_id, participant, message_type, content, context, references_message_id) VALUES ($1, $2, $3, $4, $5, $6) RETURNING * `; const messageResult = await executeQuery(messageQuery, [ conversation.id, validatedArgs.participant, validatedArgs.message_type, validatedArgs.content, JSON.stringify(validatedArgs.context), validatedArgs.references_message_id || null ]); const message = messageResult.rows[0]; // Get recent conversation context const recentQuery = ` SELECT participant, message_type, content, created_at FROM ai_conversation_messages WHERE conversation_id = $1 ORDER BY created_at DESC LIMIT 5 `; const recentResult = await executeQuery(recentQuery, [conversation.id]); return { success: true, message_id: message.id, conversation_title: conversation.title, participant: validatedArgs.participant, message_type: validatedArgs.message_type, content: validatedArgs.content, recent_context: recentResult.rows, message: `💬 ${validatedArgs.participant} sent ${validatedArgs.message_type} message to conversation`, formatted_output: `**${validatedArgs.participant.toUpperCase()}** [${validatedArgs.message_type}]: ${validatedArgs.content}` }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to send message: ${error.message}` }; } } }, // Tool 3: Start collaborative planning session ai_planning_session: { name: "ai_planning_session", description: "Initialize a structured planning session within an AI conversation for collaborative design and decision-making.", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "Session ID of the conversation" }, planning_objective: { type: "string", description: "Specific objective for this planning session" }, participants: { type: "array", items: { type: "string" }, description: "AI participants in the planning session" }, planning_context: { type: "object", description: "Context and requirements for planning", default: {} } }, required: ["session_id", "planning_objective", "participants"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = PlanningSessionSchema.parse(args); // Get conversation const convQuery = `SELECT id, title FROM ai_conversations WHERE session_id = $1`; const convResult = await executeQuery(convQuery, [validatedArgs.session_id]); if (convResult.rows.length === 0) { return { success: false, error: "Conversation session not found" }; } const conversation = convResult.rows[0]; // Create planning session const planningQuery = ` INSERT INTO ai_planning_sessions (conversation_id, planning_objective, participants, planning_data) VALUES ($1, $2, $3, $4) RETURNING * `; const planningResult = await executeQuery(planningQuery, [ conversation.id, validatedArgs.planning_objective, JSON.stringify(validatedArgs.participants), JSON.stringify(validatedArgs.planning_context) ]); const planning = planningResult.rows[0]; // Log planning session start const planningMessage = `🎯 **PLANNING SESSION STARTED** **Objective:** ${validatedArgs.planning_objective} **Participants:** ${validatedArgs.participants.join(', ')} **Session ID:** ${planning.id} This structured planning session will coordinate collaborative design and decision-making between the participating AIs.`; await executeQuery( `INSERT INTO ai_conversation_messages (conversation_id, participant, message_type, content) VALUES ($1, $2, $3, $4)`, [conversation.id, 'system', 'discussion', planningMessage] ); return { success: true, planning_session_id: planning.id, conversation_title: conversation.title, planning_objective: validatedArgs.planning_objective, participants: validatedArgs.participants, message: `🎯 Planning session started: "${validatedArgs.planning_objective}"`, next_steps: [ "Each participant should contribute using ai_send_message with message_type: proposal", "Use ai_task_assignment to delegate specific tasks", "Use ai_consensus_builder to reach agreements on decisions" ] }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to start planning session: ${error.message}` }; } } }, // Tool 4: Create task assignment between AIs ai_task_assignment: { name: "ai_task_assignment", description: "Assign a specific task from one AI participant to another within the conversation session.", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "Session ID of the conversation" }, assigned_to: { type: "string", description: "Which AI should receive the task", enum: ["claude", "openai", "replit"] }, assigned_by: { type: "string", description: "Which AI is assigning the task", enum: ["claude", "openai", "replit"] }, task_description: { type: "string", description: "Clear description of the task to be completed" }, task_type: { type: "string", description: "Category of task being assigned" }, parameters: { type: "object", description: "Specific parameters and requirements for the task", default: {} }, planning_session_id: { type: "integer", description: "Planning session this task is part of" } }, required: ["session_id", "assigned_to", "assigned_by", "task_description", "task_type"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = TaskAssignmentSchema.parse(args); // Get conversation const convQuery = `SELECT id, title FROM ai_conversations WHERE session_id = $1`; const convResult = await executeQuery(convQuery, [validatedArgs.session_id]); if (convResult.rows.length === 0) { return { success: false, error: "Conversation session not found" }; } const conversation = convResult.rows[0]; // Create task assignment with initial status const taskQuery = ` INSERT INTO ai_task_assignments (conversation_id, assigned_to, assigned_by, task_description, task_type, parameters, planning_session_id, status) VALUES ($1, $2, $3, $4, $5, $6, $7, 'assigned') RETURNING * `; const taskResult = await executeQuery(taskQuery, [ conversation.id, validatedArgs.assigned_to, validatedArgs.assigned_by, validatedArgs.task_description, validatedArgs.task_type, JSON.stringify(validatedArgs.parameters), validatedArgs.planning_session_id || null ]); const task = taskResult.rows[0]; // Send task assignment message const assignmentMessage = `📋 **TASK ASSIGNMENT** **From:** ${validatedArgs.assigned_by} **To:** ${validatedArgs.assigned_to} **Task:** ${validatedArgs.task_description} **Type:** ${validatedArgs.task_type} **Task ID:** ${task.id} ${validatedArgs.assigned_to}, please acknowledge this task assignment and proceed with implementation.`; await executeQuery( `INSERT INTO ai_conversation_messages (conversation_id, participant, message_type, content) VALUES ($1, $2, $3, $4)`, [conversation.id, validatedArgs.assigned_by, 'task_assignment', assignmentMessage] ); return { success: true, task_id: task.id, assigned_to: validatedArgs.assigned_to, assigned_by: validatedArgs.assigned_by, task_description: validatedArgs.task_description, task_type: validatedArgs.task_type, message: `📋 Task assigned: ${validatedArgs.assigned_by} → ${validatedArgs.assigned_to}: ${validatedArgs.task_type}`, next_steps: [ `${validatedArgs.assigned_to} should acknowledge the task`, `${validatedArgs.assigned_to} should update task progress using ai_task_update`, "Task completion should be reported back to the conversation" ] }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to assign task: ${error.message}` }; } } }, // Tool 5: Get conversation status and context get_ai_conversation_status: { name: "get_ai_conversation_status", description: "Get the current status, recent messages, and context of an AI conversation session.", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "Session ID of the conversation to check" }, include_full_history: { type: "boolean", description: "Whether to include complete message history", default: false } }, required: ["session_id"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = GetConversationStatusSchema.parse(args); // Get conversation details const convQuery = ` SELECT c.*, COUNT(m.id) as total_messages, COUNT(DISTINCT p.id) as planning_sessions, COUNT(DISTINCT t.id) as active_tasks FROM ai_conversations c LEFT JOIN ai_conversation_messages m ON c.id = m.conversation_id LEFT JOIN ai_planning_sessions p ON c.id = p.conversation_id LEFT JOIN ai_task_assignments t ON c.id = t.conversation_id AND t.status = 'assigned' WHERE c.session_id = $1 GROUP BY c.id `; const convResult = await executeQuery(convQuery, [validatedArgs.session_id]); if (convResult.rows.length === 0) { return { success: false, error: "Conversation session not found" }; } const conversation = convResult.rows[0]; // Get recent messages const messageLimit = validatedArgs.include_full_history ? 1000 : 10; const messagesQuery = ` SELECT participant, message_type, content, created_at FROM ai_conversation_messages WHERE conversation_id = $1 ORDER BY created_at DESC LIMIT $2 `; const messagesResult = await executeQuery(messagesQuery, [conversation.id, messageLimit]); // Get active tasks const tasksQuery = ` SELECT id, assigned_to, assigned_by, task_description, task_type, status, created_at FROM ai_task_assignments WHERE conversation_id = $1 AND status IN ('assigned', 'in_progress') ORDER BY created_at DESC `; const tasksResult = await executeQuery(tasksQuery, [conversation.id]); return { success: true, conversation: { session_id: conversation.session_id, title: conversation.title, objective: conversation.objective, orchestrator: conversation.orchestrator, participants: JSON.parse(conversation.participants), status: conversation.status, created_at: conversation.created_at, updated_at: conversation.updated_at }, statistics: { total_messages: parseInt(conversation.total_messages), planning_sessions: parseInt(conversation.planning_sessions), active_tasks: parseInt(conversation.active_tasks) }, recent_messages: messagesResult.rows.reverse(), // Show chronological order active_tasks: tasksResult.rows, message: `📊 Conversation "${conversation.title}" status retrieved` }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to get conversation status: ${error.message}` }; } } }, // Tool 6: Update task status and progress ai_task_update: { name: "ai_task_update", description: "Update the status and progress of an assigned AI task.", inputSchema: { type: "object", properties: { task_id: { type: "integer", description: "ID of the task to update" }, status: { type: "string", description: "New status of the task", enum: ["assigned", "in_progress", "completed", "failed", "delegated"] }, result: { type: "object", description: "Result data or progress information", default: {} }, message: { type: "string", description: "Progress update message" } }, required: ["task_id", "status"] }, handler: async (args: any) => { try { // Validate input const validatedArgs = TaskUpdateSchema.parse(args); // Update task status const updateQuery = ` UPDATE ai_task_assignments SET status = $1, result = $2, completed_at = CASE WHEN $1 = 'completed' THEN CURRENT_TIMESTAMP ELSE completed_at END WHERE id = $3 RETURNING * `; const updateResult = await executeQuery(updateQuery, [ validatedArgs.status, JSON.stringify(validatedArgs.result), validatedArgs.task_id ]); if (updateResult.rows.length === 0) { return { success: false, error: "Task not found" }; } const task = updateResult.rows[0]; // Send progress message to conversation if (validatedArgs.message) { const progressMessage = `📈 **TASK PROGRESS UPDATE** **Task ID:** ${task.id} **Assigned To:** ${task.assigned_to} **Status:** ${validatedArgs.status} **Update:** ${validatedArgs.message}`; await executeQuery( `INSERT INTO ai_conversation_messages (conversation_id, participant, message_type, content) VALUES ($1, $2, $3, $4)`, [task.conversation_id, task.assigned_to, 'discussion', progressMessage] ); } return { success: true, task_id: task.id, status: validatedArgs.status, assigned_to: task.assigned_to, message: `✅ Task ${task.id} status updated to: ${validatedArgs.status}` }; } catch (error: any) { return { success: false, error: error.message, message: `❌ Failed to update task: ${error.message}` }; } } } }; // Export tool definitions for MCP server registration export const getAICoordinationToolDefinitions = (): ToolDefinition[] => { return Object.values(aiCoordinationTools).map(tool => ({ name: tool.name, description: tool.description, inputSchema: tool.inputSchema })); }; // Export tool handlers for MCP server execution export const getAICoordinationToolHandlers = (): Record<string, ToolHandler> => { const handlers: Record<string, ToolHandler> = {}; Object.values(aiCoordinationTools).forEach(tool => { handlers[tool.name] = tool.handler; }); return handlers; };