Protocol

class rmote.protocol.Template(template)

Bases: object

A Mako-like template pre-compiled to a reusable render function.

Picklable - the instance stores only the original template string, so it can be passed as an argument to remote tool calls over the protocol. The class lives in protocol.py and is therefore available on the remote side without any extra sync step.

Usage:

tmpl = Template("Hello, ${name}!")
tmpl.render(name="Alice")          # local
await protocol(MyTool.run, tmpl)   # pass to remote tool

Syntax (Mako-like):

• ${expr} - evaluate expr and insert the string result;

  nested braces are handled correctly (e.g. ${{'k': 1}['k']})

• \${ - literal ${ (escape, no interpolation)

• % stmt - Python control-flow line (for / if / while / …)

• % endfor / % endif / % end - block terminators

• %% - literal % at the start of an output line

• ## comment - ignored

BLOCK_CONT = frozenset({'elif', 'else', 'except', 'finally'})

BLOCK_END = frozenset({'end', 'endfor', 'endif', 'endwhile', 'endwith'})

BLOCK_OPEN = frozenset({'class', 'def', 'for', 'if', 'try', 'while', 'with'})

__init__(template)

__reduce__()

Helper for pickle.

Return type:

tuple[type, tuple[Text]]

static compile(template)

Compile a Mako-like template string into a reusable render function.

Returns a callable that accepts **ctx keyword arguments and returns the rendered string. Results are cached so repeated calls with the same template string are free.

Return type:

Callable[..., Text]

render(**ctx)

Render this template with ctx as the variable namespace.

Return type:

Text

rmote.protocol.render_template(template, **ctx)

Compile and render a Mako-like template string in one step.

Return type:

Text

class rmote.protocol.Tool

Bases: object

rmote.protocol.tool_to_dict(cls)

Return type:

WSGIEnvironment[Text, Any]

rmote.protocol.tool_from_dict(data, context=None)

Return type:

type[Tool]

class rmote.protocol.BaseProtocol(reader, writer)

Bases: object

BOUNDARY = b'PROTOCOL READY\n'

COMPRESSION_THRESHOLD = 1024

MAGIC = b'RMOTE'

PACKET_HEADER = Struct('>5sIIQ')

async classmethod from_ssh(host, *, user=None, port=None, identity=None, python='python3', ssh_options=None, stderr=-3)

Bootstrap a Protocol over SSH.

Parameters:

• host (Text) – Remote host, optionally in user@host form.

• user (Text | None) – Remote username (-l). Overrides any user embedded in host.

• port (int | None) – SSH port (-p).

• identity (Text | None) – Path to an SSH identity file (-i).

• python (Text) – Python executable on the remote host.

• ssh_options (list[Text] | None) – Extra arguments inserted before the host in the ssh command (e.g. ["-o", "StrictHostKeyChecking=no"]).

• stderr (int) – Where to redirect remote stderr. Defaults to DEVNULL.

Return type:

Self

async classmethod from_stdio(logging_level=20)

Return type:

Self

async classmethod from_subprocess(process)

Return type:

Self

async read_boundary()

Return type:

None

async receive()

Return type:

tuple[Any, Flags, int]

async send(packet, flags, packet_id)

Return type:

None

async write_boundary()

Return type:

None

class rmote.protocol.Protocol(reader, writer)

Bases: BaseProtocol

async __aenter__()

Return type:

Self

async __aexit__(exc_type, exc_val, exc_tb)

Return type:

None

async __call__(tool, *args, **kwargs)

Overloads:

• self, tool (Callable[P, Coroutine[Any, Any, R]]), args (P.args), kwargs (P.kwargs) → R

• self, tool (Callable[P, R]), args (P.args), kwargs (P.kwargs) → R

Call self as a function.

futures: dict[int, Future[Any]]

get_id()

Return type:

int

get_log_id()

Return type:

int

tools: dict[str, Tool]

async wait_closed()

Return type:

None

class rmote.protocol.Flags(*values)

Bases: IntFlag

COMPRESSED = 1

EXCEPTION = 32

LOG = 64

REQUEST = 2

RESPONSE = 4

RPC = 16

SYNC = 8

rmote.protocol.process(*cmd_and_args, stdin=None, capture_output=False, text=False, env=None, shell=False, check=False, cwd=None)

function for execute a subprocess on the remote side, must be safe, do not share stdout/stderr of the child process, because it’s protocol pipes.

Tools must be use only this function for execute subprocesses, to avoid conflicts with protocol communication.

Return type:

CompletedProcess[Any]

rmote.protocol.bootstrap_packer(code)

Return type:

bytes