If that is the case, I can try to restructure my code and submit this feature as a pull request to your distributed module, if you think this is valuable for your rpc module.
The feature is structured in the following way:
rpc layer <-> election layer -> role dispatcher layer <-> role runner <-> wrapped highlevel apis

High level apis works like:
WORLD                               (only initialize once, tell it what roles you want to run)
\-> create_collective_group(...)    (create a sub group which supports irecv, isend, all_gather etc.)
\-> create_rpc_group(name, roles)   (create a rpc group)
\-> get_rpc_group(name, role)       (find a group handle created by a local role or remote role)

CollectiveGroup
\-> send
\-> recv
\-> isend
\-> irecv
...
\-> barrier

RpcGroup
\-> rpc_pair(key, value)           (pair a value to the group, so that it can be accessed, locally or remotely)
\-> rpc_get_paired(key)            (get a paired value in this group)
\-> rpc_sync
\-> rpc_async
\-> rpc_remote
\-> rpc_paired_class_sync          (invoke a method on the paired class instance)
\-> rpc_paired_class_async
\-> rpc_paired_class_remote
\-> rpc_paired_model_sync          (perform a forward operation on the registered model)
\-> rpc_paired_model_async
\-> rpc_paired_model_remote   

Users may start a role, register a service on this role, and access this service from another role like:
class WorkerService(object):
    # An example stateful service class
    _count = 0

    def counter(self):    
        self._count += 1
        return self._count


class Worker(RoleBase):
    def __init__(self, index):
        super(Worker, self).__init__(index)
        self.service = WorkerService()
        self.group = get_world().create_rpc_group("Employees", roles=[
            "Worker:0", "Worker:1", "Worker:2"
        ])
        # expose service
        self.group.rpc_pair("worker_service", self.service)

    def main(self):
        while True:
            pass

class Manager(RoleBase):
    def __init__(self, index):
        super(Manager, self).__init__(index)
        self.group = get_world().get_rpc_group("Employee", "Worker:{}".format(index))

    def main(self):
       for i in range(10):
            self.group.rpc_paired_class_sync(
                to="Worker:{}".format(self.role_index),
                cls_method=WorkerService.counter,
                name="worker_service",
            )

if __name__ == "__main__":
    # suppose process rank is `rank`
    # suppose there are 3 processes
    # start 3 managers and 3 workers
    world = World(world_size=3, rank=rank, 
                  roles={"Worker": (Worker, 3),
                         "Manager": (Manager, 3)},
                  rpc_timeout=0.5, election_timeout=0.3, logging=True)

What do you think about this design?