Skip to content

SiphonClient API

The SiphonClient class provides low-level gRPC communication with the Siphon server.

SiphonClient

SiphonClient

Client for the Siphon service.

Source code in eldengym/client/siphon_client.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class SiphonClient:
    """
    Client for the Siphon service.
    """

    def __init__(
        self,
        host="localhost:50051",
        max_receive_message_length=100 * 1024 * 1024,
        max_send_message_length=100 * 1024 * 1024,
    ):
        """
        Args:
            host: string, host of the server
            max_receive_message_length: int, maximum length of the receive message
            max_send_message_length: int, maximum length of the send message
        """
        self.channel = grpc.insecure_channel(
            host,
            options=[
                ("grpc.max_receive_message_length", max_receive_message_length),
                ("grpc.max_send_message_length", max_send_message_length),
            ],
        )
        self.stub = siphon_service_pb2_grpc.SiphonServiceStub(self.channel)

    def send_key(self, keys, hold_time, delay_time=0):
        """
        Send a key to the server.
        Args:
            keys: list of strings, keys to press, e.g., ['w', 'space']
            hold_time: string, time to hold the key in milliseconds
            delay_time: string, time to delay between keys in milliseconds
        """
        request = siphon_service_pb2.InputKeyTapRequest(
            keys=keys, hold_ms=hold_time, delay_ms=delay_time
        )
        return self.stub.InputKeyTap(request)

    def get_attribute(self, attributeName):
        """
        Read memory value for a single attribute.
        Args:
            attributeName: string, name of the attribute
        """
        request = siphon_service_pb2.GetSiphonRequest(attributeName=attributeName)
        response = self.stub.GetAttribute(request)

        # Handle oneof value field
        if response.HasField("int_value"):
            return response.int_value
        elif response.HasField("float_value"):
            return response.float_value
        elif response.HasField("array_value"):
            return response.array_value
        else:
            return None

    def set_attribute(self, attributeName, value):
        """
        Set the value of an attribute.
        Args:
            attributeName: string, name of the attribute
            value: int, float, or bytes - the value to set
        """
        request = siphon_service_pb2.SetSiphonRequest(attributeName=attributeName)

        # Handle oneof value field based on type
        if isinstance(value, int):
            request.int_value = value
        elif isinstance(value, float):
            request.float_value = value
        elif isinstance(value, bytes):
            request.array_value = value
        else:
            raise ValueError(
                f"Unsupported value type: {type(value)}. Must be int, float, or bytes."
            )

        return self.stub.SetAttribute(request)

    def get_frame(self):
        """
        Get current frame as numpy array.
        """
        request = siphon_service_pb2.CaptureFrameRequest()
        response = self.stub.CaptureFrame(request)
        # Server sends BGRA format (32 bits per pixel = 4 bytes per pixel)
        frame = np.frombuffer(response.frame, dtype=np.uint8)
        frame = frame.reshape(response.height, response.width, 4)  # BGRA format

        # Convert BGRA to BGR for OpenCV (remove alpha channel)
        frame = frame[:, :, :3]  # Remove alpha channel

        # save_path = os.path.join(os.path.dirname(__file__), 'frame.png')
        # cv2.imwrite(save_path, frame)
        return frame

    def move_mouse(self, delta_x, delta_y, steps=1):
        """
        Move the mouse by the given delta.
        """
        request = siphon_service_pb2.MoveMouseRequest(
            delta_x=delta_x, delta_y=delta_y, steps=steps
        )
        return self.stub.MoveMouse(request)

    def toggle_key(self, key, toggle):
        """
        Press or release a key.
        Args:
            key: string, key to toggle, e.g., 'w'
            toggle: bool, True to press, False to release
        """
        request = siphon_service_pb2.InputKeyToggleRequest(key=key, toggle=toggle)
        return self.stub.InputKeyToggle(request)

    def execute_command(
        self,
        command,
        args=None,
        working_directory="",
        timeout_seconds=30,
        capture_output=True,
    ):
        """
        Execute a command on the server.
        Args:
            command: string, command to execute
            args: list of strings, command arguments
            working_directory: string, working directory for the command
            timeout_seconds: int, timeout in seconds
            capture_output: bool, whether to capture output
        Returns:
            ExecuteCommandResponse with fields: success, message, exit_code,
            stdout_output, stderr_output, execution_time_ms
        """
        if args is None:
            args = []
        request = siphon_service_pb2.ExecuteCommandRequest(
            command=command,
            args=args,
            working_directory=working_directory,
            timeout_seconds=timeout_seconds,
            capture_output=capture_output,
        )
        return self.stub.ExecuteCommand(request)

    def set_process_config(self, process_name, process_window_name, attributes):
        """
        Set the process configuration.
        Args:
            process_name: string, name of the process
            process_window_name: string, window name of the process
            attributes: list of dicts with keys: name, pattern, offsets, type, length, method
                Example: [
                    {
                        'name': 'health',
                        'pattern': 'AB CD EF',
                        'offsets': [0x10, 0x20],
                        'type': 'int',
                        'length': 4,
                        'method': ''
                    }
                ]
        """
        request = siphon_service_pb2.SetProcessConfigRequest(
            process_name=process_name, process_window_name=process_window_name
        )

        for attr in attributes:
            proto_attr = request.attributes.add()
            proto_attr.name = attr.get("name", "")
            proto_attr.pattern = attr.get("pattern", "")
            proto_attr.offsets.extend(attr.get("offsets", []))
            proto_attr.type = attr.get("type", "")
            proto_attr.length = attr.get("length", 0)
            proto_attr.method = attr.get("method", "")

        return self.stub.SetProcessConfig(request)

    def initialize_memory(self):
        """
        Initialize the memory subsystem.
        Returns:
            InitializeMemoryResponse with fields: success, message, process_id
        """
        request = siphon_service_pb2.InitializeMemoryRequest()
        return self.stub.InitializeMemory(request)

    def initialize_input(self, window_name=""):
        """
        Initialize the input subsystem.
        Args:
            window_name: string, window name to target (optional)
        Returns:
            InitializeInputResponse with fields: success, message
        """
        request = siphon_service_pb2.InitializeInputRequest(window_name=window_name)
        return self.stub.InitializeInput(request)

    def initialize_capture(self, window_name=""):
        """
        Initialize the capture subsystem.
        Args:
            window_name: string, window name to target (optional)
        Returns:
            InitializeCaptureResponse with fields: success, message, window_width, window_height
        """
        request = siphon_service_pb2.InitializeCaptureRequest(window_name=window_name)
        return self.stub.InitializeCapture(request)

    def get_server_status(self):
        """
        Get the server initialization status.
        Returns:
            GetServerStatusResponse with fields: success, message, config_set,
            memory_initialized, input_initialized, capture_initialized,
            process_name, window_name, process_id
        """
        request = siphon_service_pb2.GetServerStatusRequest()
        return self.stub.GetServerStatus(request)

    def close(self):
        """
        Close the channel.
        """
        self.channel.close()

__init__ 

__init__(
    host="localhost:50051",
    max_receive_message_length=100 * 1024 * 1024,
    max_send_message_length=100 * 1024 * 1024,
)

Parameters:

Name Type Description Default
host

string, host of the server

 'localhost:50051'
max_receive_message_length

int, maximum length of the receive message

 100 * 1024 * 1024
max_send_message_length

int, maximum length of the send message

 100 * 1024 * 1024
Source code in eldengym/client/siphon_client.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    host="localhost:50051",
    max_receive_message_length=100 * 1024 * 1024,
    max_send_message_length=100 * 1024 * 1024,
):
    """
    Args:
        host: string, host of the server
        max_receive_message_length: int, maximum length of the receive message
        max_send_message_length: int, maximum length of the send message
    """
    self.channel = grpc.insecure_channel(
        host,
        options=[
            ("grpc.max_receive_message_length", max_receive_message_length),
            ("grpc.max_send_message_length", max_send_message_length),
        ],
    )
    self.stub = siphon_service_pb2_grpc.SiphonServiceStub(self.channel)

close 

close()

Close the channel.

Source code in eldengym/client/siphon_client.py 
229
230
231
232
233
def close(self):
    """
    Close the channel.
    """
    self.channel.close()

execute_command 

execute_command(
    command,
    args=None,
    working_directory="",
    timeout_seconds=30,
    capture_output=True,
)

Execute a command on the server. Args: command: string, command to execute args: list of strings, command arguments working_directory: string, working directory for the command timeout_seconds: int, timeout in seconds capture_output: bool, whether to capture output Returns: ExecuteCommandResponse with fields: success, message, exit_code, stdout_output, stderr_output, execution_time_ms

Source code in eldengym/client/siphon_client.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def execute_command(
    self,
    command,
    args=None,
    working_directory="",
    timeout_seconds=30,
    capture_output=True,
):
    """
    Execute a command on the server.
    Args:
        command: string, command to execute
        args: list of strings, command arguments
        working_directory: string, working directory for the command
        timeout_seconds: int, timeout in seconds
        capture_output: bool, whether to capture output
    Returns:
        ExecuteCommandResponse with fields: success, message, exit_code,
        stdout_output, stderr_output, execution_time_ms
    """
    if args is None:
        args = []
    request = siphon_service_pb2.ExecuteCommandRequest(
        command=command,
        args=args,
        working_directory=working_directory,
        timeout_seconds=timeout_seconds,
        capture_output=capture_output,
    )
    return self.stub.ExecuteCommand(request)

get_attribute 

get_attribute(attributeName)

Read memory value for a single attribute. Args: attributeName: string, name of the attribute

Source code in eldengym/client/siphon_client.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_attribute(self, attributeName):
    """
    Read memory value for a single attribute.
    Args:
        attributeName: string, name of the attribute
    """
    request = siphon_service_pb2.GetSiphonRequest(attributeName=attributeName)
    response = self.stub.GetAttribute(request)

    # Handle oneof value field
    if response.HasField("int_value"):
        return response.int_value
    elif response.HasField("float_value"):
        return response.float_value
    elif response.HasField("array_value"):
        return response.array_value
    else:
        return None

get_frame 

get_frame()

Get current frame as numpy array.

Source code in eldengym/client/siphon_client.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_frame(self):
    """
    Get current frame as numpy array.
    """
    request = siphon_service_pb2.CaptureFrameRequest()
    response = self.stub.CaptureFrame(request)
    # Server sends BGRA format (32 bits per pixel = 4 bytes per pixel)
    frame = np.frombuffer(response.frame, dtype=np.uint8)
    frame = frame.reshape(response.height, response.width, 4)  # BGRA format

    # Convert BGRA to BGR for OpenCV (remove alpha channel)
    frame = frame[:, :, :3]  # Remove alpha channel

    # save_path = os.path.join(os.path.dirname(__file__), 'frame.png')
    # cv2.imwrite(save_path, frame)
    return frame

get_server_status 

get_server_status()

Get the server initialization status. Returns: GetServerStatusResponse with fields: success, message, config_set, memory_initialized, input_initialized, capture_initialized, process_name, window_name, process_id

Source code in eldengym/client/siphon_client.py 
218
219
220
221
222
223
224
225
226
227
def get_server_status(self):
    """
    Get the server initialization status.
    Returns:
        GetServerStatusResponse with fields: success, message, config_set,
        memory_initialized, input_initialized, capture_initialized,
        process_name, window_name, process_id
    """
    request = siphon_service_pb2.GetServerStatusRequest()
    return self.stub.GetServerStatus(request)

initialize_capture 

initialize_capture(window_name='')

Initialize the capture subsystem. Args: window_name: string, window name to target (optional) Returns: InitializeCaptureResponse with fields: success, message, window_width, window_height

Source code in eldengym/client/siphon_client.py 
207
208
209
210
211
212
213
214
215
216
def initialize_capture(self, window_name=""):
    """
    Initialize the capture subsystem.
    Args:
        window_name: string, window name to target (optional)
    Returns:
        InitializeCaptureResponse with fields: success, message, window_width, window_height
    """
    request = siphon_service_pb2.InitializeCaptureRequest(window_name=window_name)
    return self.stub.InitializeCapture(request)

initialize_input 

initialize_input(window_name='')

Initialize the input subsystem. Args: window_name: string, window name to target (optional) Returns: InitializeInputResponse with fields: success, message

Source code in eldengym/client/siphon_client.py 
196
197
198
199
200
201
202
203
204
205
def initialize_input(self, window_name=""):
    """
    Initialize the input subsystem.
    Args:
        window_name: string, window name to target (optional)
    Returns:
        InitializeInputResponse with fields: success, message
    """
    request = siphon_service_pb2.InitializeInputRequest(window_name=window_name)
    return self.stub.InitializeInput(request)

initialize_memory 

initialize_memory()

Initialize the memory subsystem. Returns: InitializeMemoryResponse with fields: success, message, process_id

Source code in eldengym/client/siphon_client.py 
187
188
189
190
191
192
193
194
def initialize_memory(self):
    """
    Initialize the memory subsystem.
    Returns:
        InitializeMemoryResponse with fields: success, message, process_id
    """
    request = siphon_service_pb2.InitializeMemoryRequest()
    return self.stub.InitializeMemory(request)

move_mouse 

move_mouse(delta_x, delta_y, steps=1)

Move the mouse by the given delta.

Source code in eldengym/client/siphon_client.py 
104
105
106
107
108
109
110
111
def move_mouse(self, delta_x, delta_y, steps=1):
    """
    Move the mouse by the given delta.
    """
    request = siphon_service_pb2.MoveMouseRequest(
        delta_x=delta_x, delta_y=delta_y, steps=steps
    )
    return self.stub.MoveMouse(request)

send_key 

send_key(keys, hold_time, delay_time=0)

Send a key to the server. Args: keys: list of strings, keys to press, e.g., ['w', 'space'] hold_time: string, time to hold the key in milliseconds delay_time: string, time to delay between keys in milliseconds

Source code in eldengym/client/siphon_client.py 
32
33
34
35
36
37
38
39
40
41
42
43
def send_key(self, keys, hold_time, delay_time=0):
    """
    Send a key to the server.
    Args:
        keys: list of strings, keys to press, e.g., ['w', 'space']
        hold_time: string, time to hold the key in milliseconds
        delay_time: string, time to delay between keys in milliseconds
    """
    request = siphon_service_pb2.InputKeyTapRequest(
        keys=keys, hold_ms=hold_time, delay_ms=delay_time
    )
    return self.stub.InputKeyTap(request)

set_attribute 

set_attribute(attributeName, value)

Set the value of an attribute. Args: attributeName: string, name of the attribute value: int, float, or bytes - the value to set

Source code in eldengym/client/siphon_client.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def set_attribute(self, attributeName, value):
    """
    Set the value of an attribute.
    Args:
        attributeName: string, name of the attribute
        value: int, float, or bytes - the value to set
    """
    request = siphon_service_pb2.SetSiphonRequest(attributeName=attributeName)

    # Handle oneof value field based on type
    if isinstance(value, int):
        request.int_value = value
    elif isinstance(value, float):
        request.float_value = value
    elif isinstance(value, bytes):
        request.array_value = value
    else:
        raise ValueError(
            f"Unsupported value type: {type(value)}. Must be int, float, or bytes."
        )

    return self.stub.SetAttribute(request)

set_process_config 

set_process_config(
    process_name, process_window_name, attributes
)

Set the process configuration. Args: process_name: string, name of the process process_window_name: string, window name of the process attributes: list of dicts with keys: name, pattern, offsets, type, length, method Example: [ { 'name': 'health', 'pattern': 'AB CD EF', 'offsets': [0x10, 0x20], 'type': 'int', 'length': 4, 'method': '' } ]

Source code in eldengym/client/siphon_client.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def set_process_config(self, process_name, process_window_name, attributes):
    """
    Set the process configuration.
    Args:
        process_name: string, name of the process
        process_window_name: string, window name of the process
        attributes: list of dicts with keys: name, pattern, offsets, type, length, method
            Example: [
                {
                    'name': 'health',
                    'pattern': 'AB CD EF',
                    'offsets': [0x10, 0x20],
                    'type': 'int',
                    'length': 4,
                    'method': ''
                }
            ]
    """
    request = siphon_service_pb2.SetProcessConfigRequest(
        process_name=process_name, process_window_name=process_window_name
    )

    for attr in attributes:
        proto_attr = request.attributes.add()
        proto_attr.name = attr.get("name", "")
        proto_attr.pattern = attr.get("pattern", "")
        proto_attr.offsets.extend(attr.get("offsets", []))
        proto_attr.type = attr.get("type", "")
        proto_attr.length = attr.get("length", 0)
        proto_attr.method = attr.get("method", "")

    return self.stub.SetProcessConfig(request)

toggle_key 

toggle_key(key, toggle)

Press or release a key. Args: key: string, key to toggle, e.g., 'w' toggle: bool, True to press, False to release

Source code in eldengym/client/siphon_client.py 
113
114
115
116
117
118
119
120
121
def toggle_key(self, key, toggle):
    """
    Press or release a key.
    Args:
        key: string, key to toggle, e.g., 'w'
        toggle: bool, True to press, False to release
    """
    request = siphon_service_pb2.InputKeyToggleRequest(key=key, toggle=toggle)
    return self.stub.InputKeyToggle(request)

Core Methods

Input Control

send_key(keys, hold_time, delay_time=0)

Send keyboard input to the game.

Args: - keys (list[str]): Keys to press (e.g., ['W', 'SPACE']) - hold_time (int): Time to hold keys in milliseconds - delay_time (int): Delay between keys in milliseconds

Example: 

# Move forward for 500ms
client.send_key(['W'], 500)

# Jump (quick tap)
client.send_key(['SPACE'], 100)

# Multiple keys with delay
client.send_key(['W', 'SHIFT', 'SPACE'], 200, 100)

toggle_key(key, toggle)

Press or release a key.

Args: - key (str): Key to toggle - toggle (bool): True to press, False to release

# Hold forward
client.toggle_key('W', True)
# ... do something ...
client.toggle_key('W', False)

move_mouse(delta_x, delta_y, steps=1)

Move the mouse cursor.

Args: - delta_x (int): Horizontal movement - delta_y (int): Vertical movement - steps (int): Number of steps to interpolate

# Look right
client.move_mouse(100, 0)

# Look down
client.move_mouse(0, 50)

Memory Operations

get_attribute(attribute_name)

Read a memory value.

Args: - attribute_name (str): Name of the attribute (from config)

Returns: - int | float | bytes: The attribute value

hp = client.get_attribute("HeroHp")
max_hp = client.get_attribute("HeroMaxHp")
print(f"HP: {hp}/{max_hp}")

set_attribute(attribute_name, value)

Write a memory value.

Args: - attribute_name (str): Name of the attribute - value (int | float | bytes): Value to write

# Set player HP
client.set_attribute("HeroHp", 1000)

# Set game speed
client.set_attribute("gameSpeedVal", 2.0)

Frame Capture

get_frame()

Capture the current game frame.

Returns: - np.ndarray: BGR frame (H, W, 3), uint8

frame = client.get_frame()
print(f"Frame shape: {frame.shape}")  # e.g., (1080, 1920, 3)

Initialization

set_process_config(process_name, process_window_name, attributes)

Configure the target process and memory attributes.

Args: - process_name (str): Process name (e.g., "eldenring.exe") - process_window_name (str): Window name (e.g., "ELDEN RING") - attributes (list[dict]): Memory attribute configurations

attributes = [
    {
        'name': 'HeroHp',
        'pattern': '48 8B 05 ?? ?? ?? ??',
        'offsets': [0x10EF8, 0x0, 0x190],
        'type': 'int',
        'length': 4,
        'method': ''
    }
]

client.set_process_config("eldenring.exe", "ELDEN RING", attributes)

initialize_memory()

Initialize the memory subsystem.

Returns: - InitializeMemoryResponse: Contains success, message, process_id

initialize_input(window_name="")

Initialize the input subsystem.

Args: - window_name (str): Target window name (optional)

initialize_capture(window_name="")

Initialize the capture subsystem.

Args: - window_name (str): Target window name (optional)

Returns: - InitializeCaptureResponse: Contains success, message, window_width, window_height

get_server_status()

Get current server initialization status.

Returns: - GetServerStatusResponse: Server state information

status = client.get_server_status()
print(f"Memory initialized: {status.memory_initialized}")
print(f"Process ID: {status.process_id}")

System Commands

execute_command(command, args=None, working_directory="", timeout_seconds=30, capture_output=True)

Execute a system command on the server.

Args: - command (str): Command to execute - args (list[str]): Command arguments - working_directory (str): Working directory - timeout_seconds (int): Command timeout - capture_output (bool): Whether to capture output

Returns: - ExecuteCommandResponse: Contains success, message, exit_code, stdout_output, stderr_output

# Start the game
response = client.execute_command(
    "eldenring.exe",
    working_directory="C:/Games/Elden Ring"
)
print(f"Exit code: {response.exit_code}")

Connection

close()

Close the gRPC connection.

client.close()

Connection Parameters

from eldengym.client.siphon_client import SiphonClient

client = SiphonClient(
    host="localhost:50051",              # Server address
    max_receive_message_length=100*1024*1024,  # 100MB
    max_send_message_length=100*1024*1024,     # 100MB
)

Usage Notes

EldenClient vs SiphonClient

For Elden Ring development, use EldenClient which inherits from SiphonClient and provides game-specific helpers. Use SiphonClient directly only for non-Elden Ring applications.

Memory Operations

Direct memory operations (get_attribute, set_attribute) require proper initialization. Always call initialization methods first.