Compare commits
1 Commits
121600194c
...
46f98671ea
Author | SHA1 | Date |
---|---|---|
|
46f98671ea |
|
@ -0,0 +1 @@
|
||||||
|
.venv
|
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
@ -1,4 +1,15 @@
|
||||||
"""MCP SSH Server implementation."""
|
"""MCP SSH Server implementation.
|
||||||
|
|
||||||
|
This module implements an Anthropic Model Context Protocol (MCP) server that provides
|
||||||
|
secure SSH access to remote systems. It allows LLMs to connect to and execute commands
|
||||||
|
on remote systems via SSH, while enforcing security by requiring hostname, username,
|
||||||
|
and key file to be set in configuration rather than allowing the LLM to specify them directly.
|
||||||
|
|
||||||
|
The server provides three main tools:
|
||||||
|
1. ssh_connect: Connect to an SSH server using credentials from configuration
|
||||||
|
2. ssh_execute: Execute a command on the connected SSH server
|
||||||
|
3. ssh_disconnect: Disconnect from the SSH server
|
||||||
|
"""
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
@ -89,20 +100,25 @@ class SSHSession:
|
||||||
class SSHConnectionParams(BaseModel):
|
class SSHConnectionParams(BaseModel):
|
||||||
"""Parameters for SSH connection."""
|
"""Parameters for SSH connection."""
|
||||||
|
|
||||||
hostname: Optional[str] = Field(None, description="SSH server hostname or IP address")
|
hostname: Optional[str] = Field(None, description="SSH server hostname or IP address (ignored - set from configuration)")
|
||||||
port: Optional[int] = Field(None, description="SSH server port")
|
port: Optional[int] = Field(None, description="SSH server port (optional)")
|
||||||
username: Optional[str] = Field(None, description="SSH username")
|
username: Optional[str] = Field(None, description="SSH username (ignored - set from configuration)")
|
||||||
key_filename: Optional[str] = Field(None, description="Path to SSH private key file")
|
key_filename: Optional[str] = Field(None, description="Path to SSH private key file (ignored - set from configuration)")
|
||||||
|
|
||||||
|
|
||||||
class CommandParams(BaseModel):
|
class CommandParams(BaseModel):
|
||||||
"""Parameters for executing a command."""
|
"""Parameters for executing a command on the remote SSH server."""
|
||||||
|
|
||||||
command: str = Field(..., description="Command to execute on remote server")
|
command: str = Field(..., description="Command to execute on the remote SSH server (requires an active SSH connection)")
|
||||||
|
|
||||||
|
|
||||||
class SSHServerMCP(FastMCP):
|
class SSHServerMCP(FastMCP):
|
||||||
"""MCP server that provides SSH access to remote systems."""
|
"""MCP server that provides secure SSH access to remote systems.
|
||||||
|
|
||||||
|
This server allows LLMs to connect to and execute commands on remote systems via SSH,
|
||||||
|
while enforcing security by requiring hostname, username, and key file to be set
|
||||||
|
in configuration rather than allowing the LLM to specify them directly.
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, hostname=None, port=None, username=None, key_filename=None, server_name=None, tool_prefix=None):
|
def __init__(self, hostname=None, port=None, username=None, key_filename=None, server_name=None, tool_prefix=None):
|
||||||
"""Initialize SSH server.
|
"""Initialize SSH server.
|
||||||
|
@ -134,15 +150,16 @@ class SSHServerMCP(FastMCP):
|
||||||
execute_name = f"{self.tool_prefix}ssh_execute"
|
execute_name = f"{self.tool_prefix}ssh_execute"
|
||||||
disconnect_name = f"{self.tool_prefix}ssh_disconnect"
|
disconnect_name = f"{self.tool_prefix}ssh_disconnect"
|
||||||
|
|
||||||
self.add_tool(self.ssh_connect, name=connect_name, description=f"Connect to SSH server: {server_name}")
|
self.add_tool(self.ssh_connect, name=connect_name, description=f"Connect to SSH server: {server_name} (hostname and username are set in configuration and cannot be altered)")
|
||||||
self.add_tool(self.ssh_execute, name=execute_name, description=f"Execute a command on SSH server: {server_name}")
|
self.add_tool(self.ssh_execute, name=execute_name, description=f"Execute a command on SSH server: {server_name} using the established SSH connection")
|
||||||
self.add_tool(self.ssh_disconnect, name=disconnect_name, description=f"Disconnect from SSH server: {server_name}")
|
self.add_tool(self.ssh_disconnect, name=disconnect_name, description=f"Disconnect from SSH server: {server_name} and close the SSH connection")
|
||||||
|
|
||||||
def ssh_connect(self, params: SSHConnectionParams) -> Dict[str, Any]:
|
def ssh_connect(self, params: SSHConnectionParams) -> Dict[str, Any]:
|
||||||
"""Connect to an SSH server.
|
"""Connect to an SSH server using SSH protocol.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
params: SSH connection parameters (ignored for security-critical fields)
|
params: SSH connection parameters (hostname, username, and key_filename are ignored
|
||||||
|
and set from configuration for security reasons; only port can be optionally specified)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Result of connection attempt
|
Result of connection attempt
|
||||||
|
@ -176,13 +193,13 @@ class SSHServerMCP(FastMCP):
|
||||||
return {"success": False, "message": "Failed to connect to SSH server"}
|
return {"success": False, "message": "Failed to connect to SSH server"}
|
||||||
|
|
||||||
def ssh_execute(self, params: CommandParams) -> Dict[str, Any]:
|
def ssh_execute(self, params: CommandParams) -> Dict[str, Any]:
|
||||||
"""Execute a command on the SSH server.
|
"""Execute a command on the SSH server using the established SSH connection.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
params: Command parameters
|
params: Command parameters (requires an active SSH connection established via ssh_connect)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Command execution result
|
Command execution result with stdout, stderr, and exit_code
|
||||||
"""
|
"""
|
||||||
if not self.ssh_session or not self.ssh_session.connected:
|
if not self.ssh_session or not self.ssh_session.connected:
|
||||||
return {"success": False, "message": "Not connected to SSH server"}
|
return {"success": False, "message": "Not connected to SSH server"}
|
||||||
|
@ -190,10 +207,10 @@ class SSHServerMCP(FastMCP):
|
||||||
return self.ssh_session.execute_command(params.command)
|
return self.ssh_session.execute_command(params.command)
|
||||||
|
|
||||||
def ssh_disconnect(self) -> Dict[str, Any]:
|
def ssh_disconnect(self) -> Dict[str, Any]:
|
||||||
"""Disconnect from the SSH server.
|
"""Disconnect from the SSH server and close the SSH connection.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Disconnect result
|
Disconnect result indicating success or failure
|
||||||
"""
|
"""
|
||||||
if not self.ssh_session:
|
if not self.ssh_session:
|
||||||
return {"success": True, "message": "Not connected to SSH server"}
|
return {"success": True, "message": "Not connected to SSH server"}
|
||||||
|
@ -219,4 +236,4 @@ def main():
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|
Loading…
Reference in New Issue