Python API

python3 -m pip install xpipe_api

from xpipe_api import Client
 
# By default, Client() will read an access key from the file xpipe_auth on the local filesystem
# and talk to the XPipe HTTP server on localhost.
client = Client()
 
# If you want to connect to a PTB build of XPipe, you have to configure the client with that information
# as the PTB releases use a different port
client = Client(ptb=True)
 
# To connect to a remote instance with an API key, use
client = Client(token="foo", base_url = "http://servername:21721")

import asyncio
from xpipe_api import AsyncClient
 
async def main():
    client = AsyncClient()
 
    # Do your stuff async
 
if __name__ == "__main__":
    asyncio.run(main())

from xpipe_api import Client
 
client = Client()
 
# connection_query accepts glob-based filters on the category, connection name, and connection type
 
# Find all connections in the default category and all its subcategories
connections = client.connection_query(categories="Default/**")
 
# Find all connections in the default category without any subcategories
connections = client.connection_query(categories="Default/*")
 
# Find all top level connections in the default category
connections = client.connection_query(categories="Default", connections="*")
 
# Find all connections in the default category, including any sub connections
connections = client.connection_query(categories="Default", connections="**")
 
# Find all SSH config connections on the local machine
connections = client.connection_query(categories="Default", connections="Local Machine/SSH user config/*")
 
# Find only ssh config host connections in the "MyCategory" category that has the word "MyCompany" in its name
# - Searching for strings is case-insensitive here
# - This will not include subcategories of "MyCategory"
# - You can find all the available connection type names by just querying all connections and taking a look add the individual type names of the connections.
connections = client.connection_query(categories="MyCategory", types="sshConfigHost", connections="*MyCompany*")
 
# Query a specific connection by name in a category
# This will only return one element, so we can just access that
connection = client.connection_query(categories="MyCategory", connections="MyConnectionName")[0]
 
# Query the local machine connection
connection = client.connection_query(connections="Local Machine")[0]
 
# A connection reference is just a UUID string, so you can also specify it fixed
# If you have the HTTP API setting enabled in XPipe, you can copy the API UUID of a connection in its context menu
connection = "f0ec68aa-63f5-405c-b178-9a4454556d6b"

from xpipe_api import Client
 
client = Client()
 
connections = client.connection_query(categories="Default/**")
 
# Works in bulk, so it expects an array of connections
infos = client.connection_info(connections)
# Get the first info
info = infos[0]
 
# The connection UUID
uuid = info["connection"]
# This is an array containing all category hierarchy names
category = info["category"]
# This is an array containing all connection hierarchy names
name = info["name"]
# This is the connection type name that you can use in the query
type = info["type"]
# There is also some other data for internal state management, e.g. if a tunnel is running for example

from xpipe_api import Client
 
client = Client()
 
# The raw connection data. You can find the internal json representation of any existing connection
# by querying it first. You can use that information to then create your own connection json representation
data = {
    "type": "shellEnvironment",
    "commands": None,
    "host": {
      "storeId": "f0ec68aa-63f5-405c-b178-9a4454556d6b"
    },
    "shell": "pwsh",
    "elevated": False,
}
 
# The target category to put this connection into
category = "97458c07-75c0-4f9d-a06e-92d8cdf67c40"
 
client.connection_add(name="My Connection", conn_data=data, validate=True, category=category)

from xpipe_api import Client
 
client = Client()
 
connection = "f0ec68aa-63f5-405c-b178-9a4454556d6b"
 
# Opens the file browser in a specified starting directory for a connection
client.connection_browse(connection, "/etc")
 
# Opens a terminal session in a specified starting directory for a connection
client.connection_terminal(connection, "/etc")
 
# Toggles the session state for a connection. If this connection is a tunnel, then this operation will start or stop the tunnel
client.connection_toggle(connection, True)
 
# Refreshes a connection. This operation is the same as for validating a connection when creating it by attempting whether the connection is functioning.
client.connection_refresh(connection)

from xpipe_api import Client
import re
 
client = Client()
 
connection = "f0ec68aa-63f5-405c-b178-9a4454556d6b"
 
# This will start a shell session for the connection
# Note that this session is non-interactive, meaning that password prompts are not supported
shell_info = client.shell_start(connection)
 
# The shell dialect of the shell. For example cmd, powershell, bash, etc.
dialect = shell_info["shellDialect"]
# The operating system type of the system. For example, windows, linux, macos, bsd, solaris
osType = shell_info["osType"]
# The display name of the operating system. For example Windows 10 Home 22H2
osName = shell_info["osName"]
 
# Prints {'exitCode': 0, 'stdout': 'hello world', 'stderr': ''}
print(client.shell_exec(connection, "echo hello world"))
 
# Prints {'exitCode': 0, 'stdout': '<user>', 'stderr': ''}
print(client.shell_exec(connection, "echo $USER"))
 
# Prints {'exitCode': 127, 'stdout': 'invalid: command not found', 'stderr': ''}
print(client.shell_exec(connection, "invalid"))
 
# Extract ssh version from system
version_format = re.compile(r'[0-9.]+')
ret = client.shell_exec(connection, "ssh -V")
ssh_version = version_format.findall(ret["stderr"])[0]
 
# Clean up after ourselves by stopping the shell session
client.shell_stop(connection)

from xpipe_api import Client
 
client = Client()
 
connection = "f0ec68aa-63f5-405c-b178-9a4454556d6b"
 
# We need a shell for the file system
client.shell_start(connection)
 
# You are responsible for decoding files in the correct file encoding
file_bytes = client.fs_read(connection, "~/.ssh/config")
file_string = file_bytes.decode('utf-8')
 
# Writing files can be done via blobs
file_write_content = "test\nfile"
blob_id = client.fs_blob(file_write_content)
client.fs_write(connection, blob_id, "~/test_file.txt")
 
# You can do the same with script files as well
# This will create an executable script in the system's temp directory
script_write_content = "echo hello\necho world"
blob_script_id = client.fs_blob(file_write_content)
script_path = client.fs_script(connection, blob_script_id)
 
# You can then use this script for your commands
# Prints {'exitCode': 0, 'stdout': 'hello\nworld', 'stderr': ''}
print(client.shell_exec(connection, "\"" + script_path + "\""))
 
# Clean up after ourselves by stopping the shell session
client.shell_stop(connection)