nodelink

A high-level Python multiplayer library. Just works.

No presets. No opinions. You define the events, the data, and the behavior — nodelink handles the connections, state sync, and networking underneath.

pip install nodelink

Quickstart

Server:

from nodelink import Server

server = Server(port=5000)

@server.on("connect")
def on_connect(player):
    server.broadcast("joined", {"id": player.id})

@server.on("move")
def on_move(player, data):
    player.update_state({"x": data["x"], "y": data["y"]})

@server.on("disconnect")
def on_disconnect(player):
    server.broadcast("left", {"id": player.id})

@server.on_error()
def on_error(error, player):
    print(f"Error from {player.id}: {error}")

server.start()

Client:

from nodelink import Client

client = Client("localhost", 5000)

@client.on("connect")
def on_connect():
    client.send("move", {"x": 10, "y": 20})

@client.on("joined")
def on_joined(data):
    print(f"Player {data['id']} joined!")

@client.on("__state_sync__")
def on_sync(data):
    for player_id, state in data["players"].items():
        print(f"{player_id}: {state}")

@client.on_error()
def on_error(error):
    print(f"Error: {error}")

client.connect()

Core concepts

Events

Everything is event-based. Send an event from the client, handle it on the server (and vice versa) with a simple decorator.

# Client sends:
client.send("chat", {"msg": "hello!"})

# Server handles:
@server.on("chat")
def on_chat(player, data):
    server.broadcast("chat", {"from": player.id, "msg": data["msg"]})

# All clients receive:
@client.on("chat")
def on_chat(data):
    print(f"[{data['from']}]: {data['msg']}")

Player state sync

Each player has a .state dict. Update it on the server and nodelink automatically broadcasts only the changed fields to all clients at 20 ticks/sec.

@server.on("move")
def on_move(player, data):
    player.update_state({"x": data["x"], "y": data["y"]})
    # That's it — state is synced automatically

Force an immediate sync if you can't wait for the next tick:

server.sync_state()        # delta only
server.sync_state(full=True)  # full state

Sending to one player

@server.on("connect")
def on_connect(player):
    # Send something only to this player
    asyncio.ensure_future(player.send("welcome", {"id": player.id}))

Error handling

Errors in event handlers are caught and routed to your error handler instead of crashing everything.

@server.on_error()
def on_error(error, player):
    print(f"Error from {player.id}: {error}")

@client.on_error()
def on_error(error):
    print(f"Client error: {error}")

Send queue

Messages sent before the client is fully connected are automatically queued and delivered the moment the connection opens. No timing issues.

client.send("hello", {"msg": "this works even before connect() is called"})
client.connect()  # queued message is flushed on connect

API Reference

Server(host, port, tick_rate, delta_only)

Client(host, port)

Player

Built-in events

Requirements

• Python 3.8+
• websockets >= 12.0

License

MIT