interact.ts•6.01 kB
import { z } from 'zod';
import stripAnsi from 'strip-ansi';
import type { SessionManager } from './session.js';
import { wrapError } from '../../utils/errors.js';
import type { Logger } from '../../utils/logger.js';
import type { ToolResult } from '../../types/config.js';
const InteractSchema = z.object({
pid: z.number().int().positive().describe('Process ID from start_process'),
input: z.string().describe('Input to send to the process'),
timeout: z.number().positive().default(8000).describe('Maximum wait time in milliseconds'),
waitForPrompt: z.boolean().default(true).describe('Wait for REPL prompt or completion'),
});
export type InteractArgs = z.infer<typeof InteractSchema>;
/**
* Detect if output contains a REPL prompt or completion indicator
* Strips ANSI escape codes for reliable pattern matching
*/
function detectPromptOrCompletion(
output: string,
logger?: Logger
): { hasPrompt: boolean; hasError: boolean; isComplete: boolean; matchedPattern?: string } {
// Strip ANSI codes (colors, cursor positioning) for clean pattern matching
const cleanOutput = stripAnsi(output);
const lastLines = cleanOutput.slice(-200); // Check last 200 chars
// Common REPL prompts (expanded for better coverage)
const promptPatterns = [
/>>>\s*$/, // Python
/\.\.\.\s*$/, // Python continuation
/>\s*$/, // Node.js, many REPLs
/\$\s*$/, // Bash (no trailing space)
/\$ $/, // Bash with trailing space
/>$/, // PowerShell
/#\s*$/, // Root shell
/\*\s*$/, // Some REPLs
/:\s*$/, // Some interactive prompts
/In \[\d+\]:\s*$/, // IPython/Jupyter input
/Out\[\d+\]:\s*$/, // IPython/Jupyter output
/irb\(\w+\):\d+:\d+[*>]\s*$/, // Ruby IRB
/\w+>\s*$/, // SQL shells (mysql>, psql>, sqlite>)
];
const matchedPattern = promptPatterns.find(pattern => pattern.test(lastLines));
const hasPrompt = !!matchedPattern;
// Error indicators
const errorPatterns = [
/error:/i,
/exception:/i,
/traceback/i,
/syntaxerror/i,
/typeerror/i,
/referenceerror/i,
/cannot find/i,
/undefined/i,
];
const hasError = errorPatterns.some(pattern => pattern.test(cleanOutput));
// Completion indicators (process finished)
const completePatterns = [
/process exited/i,
/command not found/i,
/exit code/i,
];
const isComplete = completePatterns.some(pattern => pattern.test(cleanOutput));
// Debug logging (only when logger provided)
if (logger) {
logger.debug({
lastOutput: lastLines,
hasPrompt,
hasError,
isComplete,
matchedPattern: matchedPattern?.source,
outputLength: output.length,
}, 'Prompt detection result');
}
return { hasPrompt, hasError, isComplete, matchedPattern: matchedPattern?.source };
}
export async function interactWithProcessTool(
args: InteractArgs,
sessionManager: SessionManager,
logger: Logger
): Promise<ToolResult> {
try {
const session = sessionManager.get(args.pid);
if (!session) {
return {
content: [{
type: 'text',
text: `Error: Process not found: PID ${args.pid}`,
}],
};
}
if (session.state === 'terminated') {
return {
content: [{
type: 'text',
text: `Error: Process ${args.pid} has terminated`,
}],
};
}
// Clear buffer before sending input
session.outputBuffer = [];
// Send input (add newline for proper REPL handling)
session.ptyProcess.write(args.input + '\n');
session.state = 'running';
// Wait for output with smart detection
const startTime = Date.now();
let output = '';
if (args.waitForPrompt) {
// Poll for output with early exit
await new Promise<void>((resolve) => {
const checkInterval = setInterval(() => {
const elapsed = Date.now() - startTime;
output = session.outputBuffer.join('');
const detection = detectPromptOrCompletion(output, logger);
// Early exit conditions
if (detection.hasPrompt || detection.hasError || detection.isComplete || elapsed > args.timeout) {
clearInterval(checkInterval);
// Update session state
if (detection.hasPrompt) {
session.state = 'waiting';
} else if (detection.isComplete) {
session.state = 'terminated';
}
resolve();
}
}, 100); // Poll every 100ms
});
} else {
// Just wait for timeout
await new Promise(resolve => setTimeout(resolve, args.timeout));
output = session.outputBuffer.join('');
}
const elapsed = Date.now() - startTime;
const detection = detectPromptOrCompletion(output, logger);
// Determine status
let status = 'completed';
if (detection.hasPrompt) {
status = 'ready (waiting for input)';
} else if (detection.hasError) {
status = 'error';
} else if (elapsed >= args.timeout) {
status = 'timeout (may still be running)';
}
logger.info({
tool: 'interact_with_process',
pid: args.pid,
inputLength: args.input.length,
outputLength: output.length,
elapsed,
status,
}, 'Process interaction completed');
return {
content: [{
type: 'text',
text: `Process ${args.pid} | Status: ${status} | Time: ${elapsed}ms
${output || '[No output]'}`,
}],
};
} catch (error) {
const mcpError = wrapError(error, 'interact_with_process');
logger.error({ error: mcpError, args }, 'interact_with_process failed');
return {
content: [{
type: 'text',
text: `Error: ${mcpError.message}`,
}],
};
}
}
export const interactWithProcessToolDefinition = {
name: 'interact_with_process',
description: 'Send input to a running process and wait for output. Automatically detects REPL prompts and completion. Perfect for Python/Node REPL interactions.',
inputSchema: InteractSchema,
};