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 logging
|
||||
|
@ -89,20 +100,25 @@ class SSHSession:
|
|||
class SSHConnectionParams(BaseModel):
|
||||
"""Parameters for SSH connection."""
|
||||
|
||||
hostname: Optional[str] = Field(None, description="SSH server hostname or IP address")
|
||||
port: Optional[int] = Field(None, description="SSH server port")
|
||||
username: Optional[str] = Field(None, description="SSH username")
|
||||
key_filename: Optional[str] = Field(None, description="Path to SSH private key file")
|
||||
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 (optional)")
|
||||
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 (ignored - set from configuration)")
|
||||
|
||||
|
||||
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):
|
||||
"""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):
|
||||
"""Initialize SSH server.
|
||||
|
@ -134,15 +150,16 @@ class SSHServerMCP(FastMCP):
|
|||
execute_name = f"{self.tool_prefix}ssh_execute"
|
||||
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_execute, name=execute_name, description=f"Execute a command on SSH server: {server_name}")
|
||||
self.add_tool(self.ssh_disconnect, name=disconnect_name, description=f"Disconnect from 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} using the established SSH connection")
|
||||
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]:
|
||||
"""Connect to an SSH server.
|
||||
"""Connect to an SSH server using SSH protocol.
|
||||
|
||||
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:
|
||||
Result of connection attempt
|
||||
|
@ -176,13 +193,13 @@ class SSHServerMCP(FastMCP):
|
|||
return {"success": False, "message": "Failed to connect to SSH server"}
|
||||
|
||||
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:
|
||||
params: Command parameters
|
||||
params: Command parameters (requires an active SSH connection established via ssh_connect)
|
||||
|
||||
Returns:
|
||||
Command execution result
|
||||
Command execution result with stdout, stderr, and exit_code
|
||||
"""
|
||||
if not self.ssh_session or not self.ssh_session.connected:
|
||||
return {"success": False, "message": "Not connected to SSH server"}
|
||||
|
@ -190,10 +207,10 @@ class SSHServerMCP(FastMCP):
|
|||
return self.ssh_session.execute_command(params.command)
|
||||
|
||||
def ssh_disconnect(self) -> Dict[str, Any]:
|
||||
"""Disconnect from the SSH server.
|
||||
"""Disconnect from the SSH server and close the SSH connection.
|
||||
|
||||
Returns:
|
||||
Disconnect result
|
||||
Disconnect result indicating success or failure
|
||||
"""
|
||||
if not self.ssh_session:
|
||||
return {"success": True, "message": "Not connected to SSH server"}
|
||||
|
|
Loading…
Reference in New Issue