coinbase_agentkit.action_providers.ssh package

Submodules

coinbase_agentkit.action_providers.ssh.connection module

SSH Connection.

This module implements the SSHConnection class, which manages SSH connections to remote servers and provides functionality for executing commands.

@module ssh/connection

class coinbase_agentkit.action_providers.ssh.connection.CapturingRejectPolicy

Bases: MissingHostKeyPolicy

A host key policy that rejects unknown host keys but captures their details.

missing_host_key(client, hostname, key)

Handle an unknown host key encounter.

Parameters:

  • client – The SSHClient instance

  • hostname – The server hostname or IP

  • key – The server’s host key

Raises:

UnknownHostKeyError – Always, with host key details

class coinbase_agentkit.action_providers.ssh.connection.SSHConnection(params)

Bases: object

Manages an SSH connection to a remote server.

This class encapsulates all SSH connection functionality including establishing connections, executing commands, and managing the connection state.

connect()

Establish SSH connection using instance attributes.

Raises:
Return type:

None

connect_with_key(host, username, private_key, port=22, timeout=10, password=None)

Connect to a remote server using a private key.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • private_key (str | RSAKey) – SSH private key as string or paramiko.RSAKey object

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

  • password (Optional[str]) – Optional password for encrypted keys

Return type:

None

connect_with_key_path(host, username, private_key_path, port=22, timeout=10, password=None)

Connect to a remote server using a private key from file.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • private_key_path (str) – Path to SSH private key file

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

  • password (Optional[str]) – Optional password for encrypted keys

Raises:
Return type:

None

connect_with_password(host, username, password, port=22, timeout=10)

Connect to a remote server using password authentication.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • password (str) – SSH password

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

Return type:

None

disconnect()

Close SSH connection.

Raises:

SSHConnectionError – If disconnection fails

Return type:

None

download_file(remote_path, local_path)

Download a file from the remote server.

Parameters:

  • remote_path (str) – Path to the file on the remote server

  • local_path (str) – Destination path on the local machine

Raises:

SSHConnectionError – If connection is lost or file transfer fails

Return type:

None

execute(command, timeout=30, ignore_stderr=False)

Execute command on connected server.

Parameters:

  • command (str) – Shell command to execute

  • timeout (int) – Command execution timeout in seconds

  • ignore_stderr (bool) – If True, stderr output won’t cause exceptions

Returns:

Command output (stdout) and optionally stderr if present

Return type:

str

Raises:

SSHConnectionError – If connection is lost or command execution fails

get_connection_info()

Get information about the current connection.

Returns:

Connection information as a formatted string

Return type:

str

get_sftp_client()

Get an SFTP client from the current SSH connection.

Returns:

SFTP client object

Return type:

paramiko.SFTPClient

Raises:

SSHConnectionError – If there’s no active connection or SFTP initialization fails

is_connected()

Check if there’s an active SSH connection.

Returns:

Whether the connection is active

Return type:

bool

list_directory(remote_path)

List contents of a directory on the remote server.

Parameters:

remote_path (str) – Path to the directory on the remote server

Returns:

List of filenames in the directory

Return type:

list[str]

Raises:

SSHConnectionError – If connection is lost or directory listing fails

reset_connection()

Reset the connection state.

Return type:

None

upload_file(local_path, remote_path)

Upload a local file to the remote server.

Parameters:

  • local_path (str) – Path to the local file

  • remote_path (str) – Destination path on the remote server

Raises:

  • SSHConnectionError – If connection is lost or file transfer fails

  • FileNotFoundError – If the local file doesn’t exist

Return type:

None

exception coinbase_agentkit.action_providers.ssh.connection.SSHConnectionError

Bases: Exception

Exception raised for SSH connection errors.

class coinbase_agentkit.action_providers.ssh.connection.SSHConnectionParams(**data)

Bases: BaseModel

Validates SSH connection parameters.

classmethod check_auth_method_provided()

Wrap a classmethod, staticmethod, property or unbound function and act as a descriptor that allows us to detect decorated items from the class’ attributes.

This class’ __get__ returns the wrapped item’s __get__ result, which makes it transparent for classmethods and staticmethods.

wrapped

The decorator that has to be wrapped.

decorator_info

The decorator info.

shim

A wrapper function to wrap V1 style function.

connection_id: str
host: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

password: str | None
port: int
private_key: str | None
private_key_path: str | None
username: str
exception coinbase_agentkit.action_providers.ssh.connection.SSHKeyError

Bases: Exception

Exception raised for SSH key-related errors.

exception coinbase_agentkit.action_providers.ssh.connection.UnknownHostKeyError(message)

Bases: SSHConnectionError

Exception raised when a host key is not recognized.

This includes information to add the key using the ssh_add_host_key action.

coinbase_agentkit.action_providers.ssh.connection_pool module

SSH Connection Pool.

This module implements the SSHConnectionPool class, which manages multiple SSH connections and provides methods to create, retrieve, and close connections.

@module ssh/pool

class coinbase_agentkit.action_providers.ssh.connection_pool.SSHConnectionPool(max_connections=5)

Bases: object

Manages multiple SSH connections.

This class maintains a pool of SSH connections, limits the total number of connections, and provides methods to create, retrieve, and close connections.

clear_connection_pool()

Close all connections and clear all stored parameters.

Return type:

None

close_all_connections()

Close all active connections in the pool.

Return type:

None

close_and_remove_connection(connection_id)

Close a connection and remove it completely from the pool including parameters.

Parameters:

connection_id (str) – Unique identifier for the connection

Return type:

None

close_connection(connection_id)

Close and remove a connection from the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Return type:

SSHConnection | None

close_idle_connections()

Close any idle connections in the pool.

Returns:

Number of closed connections

Return type:

int

create_connection(params)

Create a new connection and add it to the pool.

Parameters:

params (SSHConnectionParams) – SSH connection parameters

Returns:

The newly created SSH connection

Return type:

SSHConnection

Raises:

  • SSHConnectionError – If the connection limit is reached

  • ValueError – If the connection parameters are invalid

get_connection(connection_id)

Get an existing connection from the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Returns:

The connection object

Return type:

SSHConnection

Raises:

SSHConnectionError – If the connection ID is not found or connection limit reached

get_connections()

Get all connections in the pool.

Returns:

Dictionary of all connections

Return type:

dict

has_connection(connection_id)

Check if a connection exists in the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Returns:

True if the connection exists in the pool

Return type:

bool

coinbase_agentkit.action_providers.ssh.schemas module

Schemas for ssh Action Provider.

This file contains the Pydantic schemas that define the input types for the ssh action provider’s actions.

@module ssh/schemas

class coinbase_agentkit.action_providers.ssh.schemas.AddHostKeySchema(**data)

Bases: BaseModel

Schema for ssh_add_host_key action.

host: str
key: str
key_type: str
known_hosts_file: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class coinbase_agentkit.action_providers.ssh.schemas.ConnectionStatusSchema(**data)

Bases: BaseModel

Schema for ssh_status action.

connection_id: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class coinbase_agentkit.action_providers.ssh.schemas.CustomSSHConnectionParams(**data)

Bases: SSHConnectionParams

Extended SSH connection parameters with known_hosts_file option.

known_hosts_file: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class coinbase_agentkit.action_providers.ssh.schemas.DisconnectSchema(**data)

Bases: BaseModel

Schema for ssh_disconnect action.

connection_id: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class coinbase_agentkit.action_providers.ssh.schemas.FileDownloadSchema(**data)

Bases: BaseModel

Schema for ssh_download action.

connection_id: str
local_path: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remote_path: str
class coinbase_agentkit.action_providers.ssh.schemas.FileUploadSchema(**data)

Bases: BaseModel

Schema for ssh_upload action.

connection_id: str
local_path: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remote_path: str
class coinbase_agentkit.action_providers.ssh.schemas.ListConnectionsSchema(**data)

Bases: BaseModel

Schema for list_connections action.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class coinbase_agentkit.action_providers.ssh.schemas.RemoteShellSchema(**data)

Bases: BaseModel

Schema for remote_shell action.

command: str
connection_id: str
ignore_stderr: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timeout: int
coinbase_agentkit.action_providers.ssh.schemas.SSHConnectionSchema

alias of CustomSSHConnectionParams

coinbase_agentkit.action_providers.ssh.ssh_action_provider module

SSH Action Provider.

This module implements the SSH Action Provider, which enables executing commands on remote servers via SSH, file transfers via SFTP, and connection management.

@module ssh/ssh_action_provider

class coinbase_agentkit.action_providers.ssh.ssh_action_provider.SshActionProvider(max_connections=10)

Bases: ActionProvider

SshActionProvider provides actions for SSH operations.

This provider enables connecting to remote servers via SSH and executing commands. It supports managing multiple concurrent SSH connections.

list_connections(args)

List all SSH connections in the pool.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

remote_shell(args)

Execute a command on the remote server.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_add_host_key(args)

Add an SSH host key to the known_hosts file.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_connect(args)

Establish SSH connection to remote server.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_disconnect(args)

Disconnects from an active SSH session.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_download(args)

Download a file from the remote server.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_status(args)

Retrieve status of a specific SSH connection.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

ssh_upload(args)

Upload a file to the remote server.

Parameters:

args (dict[str, Any]) – Input arguments for the action.

Returns:

A message containing the action response or error details.

Return type:

str

supports_network(network)

Check if this provider supports the specified network.

Parameters:

network (Network) – The network to check.

Returns:

Always True as SSH is network-agnostic.

Return type:

bool

coinbase_agentkit.action_providers.ssh.ssh_action_provider.ssh_action_provider(max_connections=10)

Create a new instance of the SshActionProvider.

Parameters:

max_connections (int) – Maximum number of concurrent SSH connections (default: 10)

Return type:

SshActionProvider

Returns:

An initialized SshActionProvider

Module contents

SSH connection package.

This package provides SSH connection functionality for the agent toolkit.

class coinbase_agentkit.action_providers.ssh.SSHConnection(params)

Bases: object

Manages an SSH connection to a remote server.

This class encapsulates all SSH connection functionality including establishing connections, executing commands, and managing the connection state.

connect()

Establish SSH connection using instance attributes.

Raises:
Return type:

None

connect_with_key(host, username, private_key, port=22, timeout=10, password=None)

Connect to a remote server using a private key.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • private_key (str | RSAKey) – SSH private key as string or paramiko.RSAKey object

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

  • password (Optional[str]) – Optional password for encrypted keys

Return type:

None

connect_with_key_path(host, username, private_key_path, port=22, timeout=10, password=None)

Connect to a remote server using a private key from file.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • private_key_path (str) – Path to SSH private key file

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

  • password (Optional[str]) – Optional password for encrypted keys

Raises:
Return type:

None

connect_with_password(host, username, password, port=22, timeout=10)

Connect to a remote server using password authentication.

Parameters:

  • host (str) – Remote server hostname/IP

  • username (str) – SSH username

  • password (str) – SSH password

  • port (int) – SSH port number (default: 22)

  • timeout (int) – Connection timeout in seconds (default: 10)

Return type:

None

disconnect()

Close SSH connection.

Raises:

SSHConnectionError – If disconnection fails

Return type:

None

download_file(remote_path, local_path)

Download a file from the remote server.

Parameters:

  • remote_path (str) – Path to the file on the remote server

  • local_path (str) – Destination path on the local machine

Raises:

SSHConnectionError – If connection is lost or file transfer fails

Return type:

None

execute(command, timeout=30, ignore_stderr=False)

Execute command on connected server.

Parameters:

  • command (str) – Shell command to execute

  • timeout (int) – Command execution timeout in seconds

  • ignore_stderr (bool) – If True, stderr output won’t cause exceptions

Returns:

Command output (stdout) and optionally stderr if present

Return type:

str

Raises:

SSHConnectionError – If connection is lost or command execution fails

get_connection_info()

Get information about the current connection.

Returns:

Connection information as a formatted string

Return type:

str

get_sftp_client()

Get an SFTP client from the current SSH connection.

Returns:

SFTP client object

Return type:

paramiko.SFTPClient

Raises:

SSHConnectionError – If there’s no active connection or SFTP initialization fails

is_connected()

Check if there’s an active SSH connection.

Returns:

Whether the connection is active

Return type:

bool

list_directory(remote_path)

List contents of a directory on the remote server.

Parameters:

remote_path (str) – Path to the directory on the remote server

Returns:

List of filenames in the directory

Return type:

list[str]

Raises:

SSHConnectionError – If connection is lost or directory listing fails

reset_connection()

Reset the connection state.

Return type:

None

upload_file(local_path, remote_path)

Upload a local file to the remote server.

Parameters:

  • local_path (str) – Path to the local file

  • remote_path (str) – Destination path on the remote server

Raises:

  • SSHConnectionError – If connection is lost or file transfer fails

  • FileNotFoundError – If the local file doesn’t exist

Return type:

None

exception coinbase_agentkit.action_providers.ssh.SSHConnectionError

Bases: Exception

Exception raised for SSH connection errors.

class coinbase_agentkit.action_providers.ssh.SSHConnectionParams(**data)

Bases: BaseModel

Validates SSH connection parameters.

classmethod check_auth_method_provided()

Wrap a classmethod, staticmethod, property or unbound function and act as a descriptor that allows us to detect decorated items from the class’ attributes.

This class’ __get__ returns the wrapped item’s __get__ result, which makes it transparent for classmethods and staticmethods.

wrapped

The decorator that has to be wrapped.

decorator_info

The decorator info.

shim

A wrapper function to wrap V1 style function.

connection_id: str
host: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

password: str | None
port: int
private_key: str | None
private_key_path: str | None
username: str
class coinbase_agentkit.action_providers.ssh.SSHConnectionPool(max_connections=5)

Bases: object

Manages multiple SSH connections.

This class maintains a pool of SSH connections, limits the total number of connections, and provides methods to create, retrieve, and close connections.

clear_connection_pool()

Close all connections and clear all stored parameters.

Return type:

None

close_all_connections()

Close all active connections in the pool.

Return type:

None

close_and_remove_connection(connection_id)

Close a connection and remove it completely from the pool including parameters.

Parameters:

connection_id (str) – Unique identifier for the connection

Return type:

None

close_connection(connection_id)

Close and remove a connection from the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Return type:

SSHConnection | None

close_idle_connections()

Close any idle connections in the pool.

Returns:

Number of closed connections

Return type:

int

create_connection(params)

Create a new connection and add it to the pool.

Parameters:

params (SSHConnectionParams) – SSH connection parameters

Returns:

The newly created SSH connection

Return type:

SSHConnection

Raises:

  • SSHConnectionError – If the connection limit is reached

  • ValueError – If the connection parameters are invalid

get_connection(connection_id)

Get an existing connection from the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Returns:

The connection object

Return type:

SSHConnection

Raises:

SSHConnectionError – If the connection ID is not found or connection limit reached

get_connections()

Get all connections in the pool.

Returns:

Dictionary of all connections

Return type:

dict

has_connection(connection_id)

Check if a connection exists in the pool.

Parameters:

connection_id (str) – Unique identifier for the connection

Returns:

True if the connection exists in the pool

Return type:

bool

exception coinbase_agentkit.action_providers.ssh.SSHKeyError

Bases: Exception

Exception raised for SSH key-related errors.