ssh_execute
Execute commands on remote systems via established SSH sessions, managing inputs and timeouts for efficient remote task execution through the SSH MCP Server interface.
Instructions
SSH 세션에서 명령 실행
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | ||
| session_id | Yes | ||
| timeout | No |
Implementation Reference
- main.py:245-267 (handler)The handler function for the 'ssh_execute' MCP tool. It is registered via @mcp.tool() decorator and handles executing commands on an existing SSH session, formatting and returning the output.
@mcp.tool() async def ssh_execute(session_id: str, command: str, timeout: int = 30) -> str: """SSH 세션에서 명령 실행""" try: result = await ssh_manager.execute_command(session_id, command, timeout) output = [] output.append(f"Session: {session_id}") output.append(f"Command: {command}") output.append(f"Exit Status: {result['exit_status']}") if result['stdout']: output.append("STDOUT:") output.append(result['stdout']) if result['stderr']: output.append("STDERR:") output.append(result['stderr']) return "\n".join(output) except Exception as e: return f"Command execution failed: {str(e)}" - main.py:72-106 (helper)Supporting method in SSHSessionManager class that performs the core SSH command execution using asyncssh, handles timeouts, updates session metadata, and returns structured results.
async def execute_command(self, session_id: str, command: str, timeout: int = 30) -> Dict[str, Any]: """세션에서 명령 실행""" if session_id not in self.connections: raise Exception(f"Session '{session_id}' not found") conn = self.connections[session_id] metadata = self.session_metadata[session_id] try: # 명령 실행 result = await asyncio.wait_for( conn.run(command, check=False), timeout=timeout ) # 메타데이터 업데이트 metadata['last_used'] = time.time() metadata['command_count'] += 1 return { 'session_id': session_id, 'command': command, 'stdout': result.stdout, 'stderr': result.stderr, 'exit_status': result.exit_status, 'execution_time': time.time() } except asyncio.TimeoutError: raise Exception(f"Command timeout after {timeout} seconds") except Exception as e: logger.error(f"Command execution failed in session {session_id}: {e}") raise Exception(f"Command execution failed: {str(e)}")