diff --git a/examples/pyxsim/example_pyxsim.py b/examples/pyxsim/example_pyxsim.py
index cfd6e21..0960661 100644
--- a/examples/pyxsim/example_pyxsim.py
+++ b/examples/pyxsim/example_pyxsim.py
@@ -1,5 +1,6 @@
 # Copyright 2025-2026 XMOS LIMITED.
 # This Software is subject to the terms of the XMOS Public Licence: Version 1.
+import os
 from pathlib import Path
 
 import Pyxsim
@@ -20,7 +21,11 @@ def run(self) -> None:
 
 
 def main() -> None:
-    xe = Path(__file__).parent / "bin" / "example_pyxsim.xe"
+    XE_OVERRIDE = os.environ.get("XE_FILE_OVERRIDE")
+    if XE_OVERRIDE:
+        xe = Path(XE_OVERRIDE)
+    else:
+        xe = Path(__file__).parent / "bin" / "example_pyxsim.xe"
     assert xe.exists()
 
     Pyxsim.run_with_pyxsim(
diff --git a/lib/python/Pyxsim/__init__.py b/lib/python/Pyxsim/__init__.py
index 756ad1f..e5071fb 100644
--- a/lib/python/Pyxsim/__init__.py
+++ b/lib/python/Pyxsim/__init__.py
@@ -17,6 +17,7 @@
 from . import pyxsim
 from Pyxsim.xe import Xe
 from Pyxsim.pyxsim import Xsi
+from Pyxsim.pyxsim import XsiRemote
 
 
 # This function is called automatically by the runners
@@ -195,7 +196,12 @@ def run_on_simulator(*args, **kwargs):
 
 
 def do_run_pyxsim(xe, simargs, appargs, simthreads, plugins=None):
-    xsi = pyxsim.Xsi(xe_path=xe, simargs=simargs, appargs=appargs)
+    # Get XSI_ENDPOINT environment variable if set, otherwise use local
+    xsi_endpoint = os.environ.get("XSI_ENDPOINT")
+    if xsi_endpoint:
+        xsi = pyxsim.XsiRemote(xsi_endpoint, xe_path=xe, simargs=simargs, appargs=appargs)
+    else:
+        xsi = pyxsim.Xsi(xe_path=xe, simargs=simargs, appargs=appargs) 
     for x in simthreads:
         xsi.register_simthread(x)
     if plugins:
@@ -257,6 +263,9 @@ def run_with_pyxsim(
 
         simargs += ["--vcd-tracing", vcd_args]
 
+    if os.environ.get("XSI_ENDPOINT"):
+        timeout = None  # Don't timeout when running on remote server
+
     p = ctx.Process(
         target=do_run_pyxsim, args=(xe_path, simargs, appargs, simthreads, plugins)
     )
diff --git a/lib/python/Pyxsim/pyxsim.py b/lib/python/Pyxsim/pyxsim.py
index b95d867..52d8889 100644
--- a/lib/python/Pyxsim/pyxsim.py
+++ b/lib/python/Pyxsim/pyxsim.py
@@ -243,52 +243,44 @@ def run(self):
         self.resume_condition = lambda x: False
         self.complete_event.set()
 
-
-class Xsi:
+class XsiBase:
     @staticmethod
     def get_xsi_tick_freq_hz():
         """
         Returns the tick frequency corresponding to the time resolution used in xsim
         """
-        xsi_tick_freq_hz = float(1e15) # Time resolution used within xsim (1 femtosecond)
-        return xsi_tick_freq_hz
+        return float(1e15)  # Time resolution used within xsim (1 femtosecond)
 
     @staticmethod
     def get_xsi_tick_freq_khz():
         """
         Returns the tick frequency used in xsim in KHz
         """
-        return Xsi.get_xsi_tick_freq_hz()/1000.0
+        return XsiBase.get_xsi_tick_freq_hz() / 1000.0
 
     @staticmethod
     def get_xsi_tick_freq_mhz():
         """
         Returns the tick frequency used in xsim in MHz
         """
-        return Xsi.get_xsi_tick_freq_hz()/1000000.0
+        return XsiBase.get_xsi_tick_freq_hz() / 1000000.0
 
     def __init__(self, xe_path=None, simargs=[], appargs=[]):
-        self.xsim = c_void_p()
         self.xe_path = xe_path
-        args = " ".join(['"{}"'.format(x) for x in simargs + [self.xe_path] + appargs])
-        if platform_is_windows():
-            args = args.replace("\\", "/")
-        c_args = c_char_p(args.encode("utf-8"))
-        xsi_lib.xsi_create(byref(self.xsim), c_args)
         self._plugins = []
         self._simthreads = []
         self._time = 0
         self.xe = Xe(self.xe_path)
-        self._time_step = Xsi.get_xsi_tick_freq_mhz() / self.xe.freq  # time-step in fs
+        self._time_step = self.get_xsi_tick_freq_mhz() / self.xe.freq
 
     def register_plugin(self, plugin):
         self._plugins.append(plugin)
 
     def register_simthread(self, fn):
         if isinstance(fn, tuple):
-            xs = list(fn)
-            fn = xs[0]
-            args = xs[1:]
+            simthread_args = list(fn)
+            fn = simthread_args[0]
+            args = simthread_args[1:]
         else:
             args = []
 
@@ -304,9 +296,8 @@ def wait_for_simthread(simthread):
             raise TestError("Simthread encoutered an exception")
 
     def clock(self):
-        status = xsi_lib.xsi_clock(self.xsim)
-        # time + (time_resolution/xe_freq_hz) = (time*xe_freq_hz + time_resolution)/xe_freq_hz
-        self._time = ((self._time * self.xe.freq) + (Xsi.get_xsi_tick_freq_mhz())) / self.xe.freq
+        status = self._clock()
+        self._time += self._time_step
         if XsiStatus.is_valid(status):
             for plugin in self._plugins:
                 plugin.clock(self)
@@ -329,14 +320,13 @@ def get_time_ns(self):
         """
         Returns current time in nanoseconds
         """
-        return (self._time * 1e9) / (Xsi.get_xsi_tick_freq_hz())
+        return (self._time * 1e9) / self.get_xsi_tick_freq_hz()
 
     def get_time_us(self):
         """
         Returns current time in microseconds
         """
-        return (self._time * 1e6) / (Xsi.get_xsi_tick_freq_hz())
-
+        return (self._time * 1e6) / self.get_xsi_tick_freq_hz()
 
     def run(self):
         status = XsiStatus.OK
@@ -347,6 +337,80 @@ def run(self):
         while status != XsiStatus.DONE:
             status = self.clock()
             XsiStatus.error_if_not_valid(status)
+    
+    def read_symbol_word(self, tile, symbol, offset=0):
+        address = self.xe.symtab[tile, symbol] + offset
+        buf = self.read_mem(tile, address, 4, return_ctype=True)
+        return struct.unpack("<I", buf)
+
+    def read_symbol_byte(self, tile, symbol, offset=0):
+        address = self.xe.symtab[tile, symbol] + offset
+        buf = self.read_mem(tile, address, 1, return_ctype=True)
+        return ord(buf[0])
+    
+    def write_symbol_word(self, tile, symbol, value, offset=0):
+        address = self.xe.symtab[tile, symbol] + offset
+        self.write_mem(tile, address, 4, struct.pack("<I", value))
+
+    def write_symbol_byte(self, tile, symbol, value, offset=0):
+        address = self.xe.symtab[tile, symbol] + offset
+        self.write_mem(tile, address, 1, struct.pack("<c", value))
+
+    def _clock(self):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def terminate(self):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def sample_pin(self, package, pin):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def sample_port_pins(self, tile, port, mask):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def drive_pin(self, package, pin, value):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def drive_port_pins(self, tile, port, mask, value):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def drive_periph_pin(self, periph, pin, mask, value):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def sample_periph_pin(self, periph, pin, mask):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def is_pin_driving(self, package, pin):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def is_port_pins_driving(self, tile, port):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def read_mem(self, tile, address, num_bytes, return_ctype=False):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def write_mem(self, tile, address, num_bytes, data):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def read_pswitch_reg(self, tile, reg_num):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+    def write_pswitch_reg(self, tile, reg_num, value):
+        raise NotImplementedError("XsiBase is an abstract class and cannot be instantiated directly")
+
+
+class Xsi(XsiBase):
+    def __init__(self, xe_path=None, simargs=[], appargs=[]):
+        super().__init__(xe_path, simargs, appargs)
+        self.xsim = c_void_p()
+        args = " ".join(['"{}"'.format(x) for x in simargs + [self.xe_path] + appargs])
+        if platform_is_windows():
+            args = args.replace("\\", "/")
+        c_args = c_char_p(args.encode("utf-8"))
+        xsi_lib.xsi_create(byref(self.xsim), c_args)
+
+    def _clock(self):
+        return xsi_lib.xsi_clock(self.xsim)
 
     def terminate(self):
         status = xsi_lib.xsi_terminate(self.xsim)
@@ -440,18 +504,6 @@ def read_mem(self, tile, address, num_bytes, return_ctype=False):
             return buf
         return list(buf)
 
-    def read_symbol_word(self, tile, symbol, offset=0):
-        address = self.xe.symtab[tile, symbol]
-        address += offset
-        buf = self.read_mem(tile, address, 4, return_ctype=True)
-        return struct.unpack("<I", buf)
-
-    def read_symbol_byte(self, tile, symbol, offset=0):
-        address = self.xe.symtab[tile, symbol]
-        address += offset
-        buf = self.read_mem(tile, address, 1, return_ctype=True)
-        return ord(buf[0])
-
     def write_mem(self, tile, address, num_bytes, data):
         c_tile = c_char_p(tile.encode("utf-8"))
         c_address = c_uint(address)
@@ -460,18 +512,6 @@ def write_mem(self, tile, address, num_bytes, data):
         status = xsi_lib.xsi_write_mem(self.xsim, c_tile, c_address, c_num_bytes, buf)
         XsiStatus.error_if_not_valid(status)
 
-    def write_symbol_word(self, tile, symbol, value, offset=0):
-        address = self.xe.symtab[tile, symbol]
-        address += offset
-        data = struct.pack("<I", value)
-        self.write_mem(tile, address, 4, data)
-
-    def write_symbol_byte(self, tile, symbol, value, offset=0):
-        address = self.xe.symtab[tile, symbol]
-        address += offset
-        data = struct.pack("<c", value)
-        self.write_mem(tile, address, 1, data)
-
     def read_pswitch_reg(self, tile, reg_num):
         c_tile = c_char_p(tile.encode("utf-8"))
         c_reg_num = c_uint(reg_num)
@@ -489,6 +529,156 @@ def write_pswitch_reg(self, tile, reg_num, value):
         status = xsi_lib.xsi_write_pswitch_reg(self.xsim, c_tile, c_reg_num, c_value)
         XsiStatus.error_if_not_valid(status)
 
+_xtc_python_lib_path = os.path.abspath(
+    os.path.join(xcc_exec_prefix, "..", "lib/python")
+)
+
+if os.path.exists(os.path.join(_xtc_python_lib_path, "xsi_remote")):
+    sys.path.append(_xtc_python_lib_path)
+    sys.path.append(os.path.join(_xtc_python_lib_path, "xsi_remote"))
+    from xsi_remote.xsi_pb2_grpc import XsiStub
+    import xsi_remote.xsi_pb2 as xsi_pb2
+    import grpc
+    class XsiRemote(XsiBase):
+        def __init__(self, server_address, xe_path=None, simargs=[], appargs=[]):
+            super().__init__(xe_path, simargs, appargs)
+            self._channel = grpc.insecure_channel(server_address)
+            self._xsi = XsiStub(self._channel)
+            # Start the sim
+            response = self._xsi.Create(xsi_pb2.CreateRequest(arguments=" ".join(simargs)))
+            XsiStatus.error_if_not_valid(response.status)
+            self._instance = response.instance_id
+
+        def _clock(self):
+            request = xsi_pb2.InstanceRequest(instance_id=self._instance)
+            response: xsi_pb2.StatusResponse = self._xsi.Clock(request)
+            return response.status
+
+        def terminate(self):
+            response = self._xsi.Terminate(xsi_pb2.InstanceRequest(instance_id=self._instance))
+            XsiStatus.error_if_not_valid(response.status)
+            self._channel.close()
+
+        def sample_pin(self, package, pin):
+            request = xsi_pb2.PinRequest(
+                instance_id=self._instance, package=package, pin=pin
+            )
+            response: xsi_pb2.ReadPinResponse = self._xsi.SamplePin(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def sample_port_pins(self, tile, port, mask):
+            # WARNING: xsi_sample_port_pins() is not always a passive observation
+            request = xsi_pb2.SamplePortPinsRequest(
+                instance_id=self._instance,
+                core=tile,
+                port=port,
+                mask=mask
+            )
+            response: xsi_pb2.ReadPortResponse = self._xsi.SamplePortPins(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def drive_pin(self, package, pin, value):
+            request = xsi_pb2.DrivePinRequest(
+                instance_id=self._instance,
+                package=package,
+                pin=pin,
+                value=value
+            )
+            response: xsi_pb2.StatusResponse = self._xsi.DrivePin(request)
+            XsiStatus.error_if_not_valid(response.status)
+
+        def drive_port_pins(self, tile, port, mask, value):
+            request = xsi_pb2.DrivePortPinsRequest(
+                instance_id=self._instance,
+                core=tile,
+                port=port,
+                mask=mask,
+                value=value,
+            )
+            response: xsi_pb2.StatusResponse = self._xsi.DrivePortPins(request)
+            XsiStatus.error_if_not_valid(response.status)
+
+        def drive_periph_pin(self, periph, pin, mask, value):
+            request = xsi_pb2.DrivePeriphPinRequest(
+                instance_id=self._instance,
+                periph=periph,
+                pin=pin,
+                mask=mask,
+                value=value,
+            )
+            response: xsi_pb2.StatusResponse = self._xsi.DrivePeriphPin(request)
+            XsiStatus.error_if_not_valid(response.status)
+
+        def sample_periph_pin(self, periph, pin, mask):
+            request = xsi_pb2.SamplePeriphPinRequest(
+                instance_id=self._instance,
+                periph=periph,
+                pin=pin,
+                mask=mask,
+            )
+            response: xsi_pb2.ReadPortResponse = self._xsi.SamplePeriphPin(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def is_pin_driving(self, package, pin):
+            request = xsi_pb2.PinRequest(
+                instance_id=self._instance, package=package, pin=pin
+            )
+            response: xsi_pb2.ReadPinResponse = self._xsi.IsPinDriving(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def is_port_pins_driving(self, tile, port):
+            request = xsi_pb2.PortRequest(
+                instance_id=self._instance, core=tile, port=port
+            )
+            response: xsi_pb2.ReadPortResponse = self._xsi.IsPortPinsDriving(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def read_mem(self, tile, address, num_bytes, return_ctype=False):
+            request = xsi_pb2.ReadMemRequest(
+                instance_id=self._instance,
+                core=tile,
+                address=address,
+                num_bytes=num_bytes,
+            )
+            response: xsi_pb2.ReadMemResponse = self._xsi.ReadMem(request)
+            XsiStatus.error_if_not_valid(response.status)
+            buf = create_string_buffer(response.data, num_bytes)
+            if return_ctype:
+                return buf
+            return list(buf)
+
+        def write_mem(self, tile, address, num_bytes, data):
+            request = xsi_pb2.WriteMemRequest(
+                instance_id=self._instance,
+                core=tile,
+                address=address,
+                data=bytes(data[:num_bytes]),
+            )
+            response: xsi_pb2.StatusResponse = self._xsi.WriteMem(request)
+            XsiStatus.error_if_not_valid(response.status)
+
+        def read_pswitch_reg(self, tile, reg_num):
+            request = xsi_pb2.ReadSwitchRegRequest(
+                instance_id=self._instance, core=tile, reg_num=reg_num
+            )
+            response: xsi_pb2.ReadRegisterResponse = self._xsi.ReadPSwitchReg(request)
+            XsiStatus.error_if_not_valid(response.status)
+            return response.value
+
+        def write_pswitch_reg(self, tile, reg_num, value):
+            request = xsi_pb2.WriteSwitchRegRequest(
+                instance_id=self._instance,
+                core=tile,
+                reg_num=reg_num,
+                value=value,
+            )
+            response: xsi_pb2.StatusResponse = self._xsi.WritePSwitchReg(request)
+            XsiStatus.error_if_not_valid(response.status)
 
 class XsiPlugin:
     def clock(self, xsi):
diff --git a/lib/python/Pyxsim/xe.py b/lib/python/Pyxsim/xe.py
index 605a34b..8647e04 100644
--- a/lib/python/Pyxsim/xe.py
+++ b/lib/python/Pyxsim/xe.py
@@ -61,6 +61,26 @@ def get_port_pins(self, port):
                     return [(package, pin, bitnum)]
             raise TestError("Cannot find port pins")
         return self._port_map[port]
+    
+    def get_elf_files(self):
+        elf_files = []
+        for root, dirs, files in os.walk(self._tempdir):
+            for f in files:
+                if f.endswith(".elf"):
+                    elf_files.append(os.path.join(root, f))
+        return elf_files
+    
+    def get_elf_file_for_tile(self, node, tile):
+        elf_files = self.get_elf_files()
+        # The elf file for a tile is called image_n0c0.elf where n0 is the node number and c0 is the tile number.
+        # But if there is a image called image_n0c0_2.elf, then that takes priority.
+        elf_file = None
+        for f in elf_files:
+            m = re.match(r".*image_n%sc%s(_\d*)?\.elf" % (node, tile), f)
+            if m:
+                if elf_file is None or m.groups(0)[0] is not None:
+                    elf_file = f
+        return elf_file
 
     def _get_symtab(self):
         stdout, _stderr = call_get_output(["xobjdump", "-t", self.path])
diff --git a/requirements.txt b/requirements.txt
index fa1914f..b026869 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -19,6 +19,8 @@
 # same modules should appear in the setup.py list as given below.
 
 colorama==0.4.6
+grpcio==1.82.1
+protobuf==7.35.1
 
 # Development dependencies
 #
diff --git a/setup.py b/setup.py
index 13d0316..349b6bd 100644
--- a/setup.py
+++ b/setup.py
@@ -15,5 +15,7 @@
     packages=setuptools.find_packages(),
     install_requires=[
         "colorama>=0.4.6",
+        "grpcio>=1.82.1",
+        "protobuf>=7.35.1",
     ],
 )