ssh_connect
Establish and manage SSH connections by creating new sessions with specified host, username, password, and port. Ideal for remote server access and command execution through the SSH MCP Server.
Instructions
SSH 서버에 연결하여 새 세션 생성
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| host | Yes | ||
| password | Yes | ||
| port | No | ||
| session_name | No | ||
| username | Yes |
Implementation Reference
- main.py:229-243 (handler)The main handler function for the 'ssh_connect' MCP tool. Decorated with @mcp.tool() for automatic registration. Creates an SSH session via SSHSessionManager and returns the session ID or error.
@mcp.tool() async def ssh_connect(host: str, username: str, password: str, port: int = 22, session_name: str = None) -> str: """SSH 서버에 연결하여 새 세션 생성""" try: session_id = await ssh_manager.create_session( host=host, username=username, password=password, port=port, session_name=session_name ) return f"SSH session created: {session_id}" except Exception as e: return f"Connection failed: {str(e)}" - main.py:26-70 (helper)Core implementation of SSH connection in SSHSessionManager.create_session(). Establishes asyncssh connection, manages session metadata, limits concurrent sessions, and starts monitoring task.
async def create_session(self, host: str, username: str, password: str, port: int = 22, session_name: Optional[str] = None) -> str: """새 SSH 세션 생성""" # 세션 수 제한 확인 if len(self.connections) >= self.max_sessions: await self._cleanup_oldest_session() # 세션 ID 생성 session_id = session_name or f"ssh_{uuid.uuid4().hex[:8]}" try: # SSH 연결 생성 conn = await asyncssh.connect( host=host, port=port, username=username, password=password, known_hosts=None, # 개발용 - 운영에서는 적절한 호스트 키 검증 필요 client_keys=None, passphrase=None ) # 세션 저장 self.connections[session_id] = conn self.session_metadata[session_id] = { 'host': host, 'port': port, 'username': username, 'created_at': time.time(), 'last_used': time.time(), 'command_count': 0 } # 연결 모니터링 태스크 시작 self.connection_tasks[session_id] = asyncio.create_task( self._monitor_session(session_id) ) logger.info(f"SSH session created: {session_id} -> {username}@{host}:{port}") return session_id except Exception as e: logger.error(f"Failed to create SSH session: {e}") raise Exception(f"SSH connection failed: {str(e)}") - main.py:229-229 (registration)The @mcp.tool() decorator registers the ssh_connect function as an MCP tool in FastMCP.
@mcp.tool()