Example: Chat Client

Example: Chat Client

This example builds an interactive chat client that reads user input from the terminal and sends/receives chat messages in real time.

View source on GitHub

Prerequisites

This example requires the aioconsole package for non-blocking terminal input:

pip install aioconsole

Why not just use Python's built-in input()? Because input() blocks the entire program while waiting for the user to type. Since the client needs to keep receiving messages from the server at the same time, we need aioconsole.ainput(), which works with asyncio and doesn't block other tasks.

What this example does

1. Connects to the server and logs in with a password
2. Filters incoming messages to only show chat messages (ignoring item sends, joins, etc.)
3. Reads user input and sends it as chat messages
4. Displays who sent each message using the player name

Code

import asyncio

import aioconsole

from archipelagopy import Client, packets, structs, enums
from archipelagopy.callback_interface import OnConnectExceptionUnion
from archipelagopy.structs import NetworkSlot


class ChatClient(Client):
    def __init__(self, port: int):
        super().__init__(port)
        self.slot_info: dict[int, NetworkSlot] = {}

    async def on_ready(self):
        print("Connected to the server.")

        await self.send(
            packets.Connect(
                version=structs.Version(major=6, minor=0, build=0),
                name="Link",  # slot name
                game="Ocarina of Time",
                password="123456",
                slot_data=True
            )
        )

    def on_connect_error(self, error: OnConnectExceptionUnion):
        print("Failed to connect:", error)

    async def on_connection_refused(self, packet: packets.ConnectionRefused):
        print(f"Connection refused: {",".join(packet.errors)}")
        await self.stop()

    async def on_connected(self, packet: packets.Connected):
        self.slot_info.update(packet.slot_info)
        print(f"Logged in as: {packet.slot_info[packet.slot].name}")

    async def on_print_json(self, packet: packets.PrintJSON):
        if packet.type != enums.PrintJSONType.CHAT:
            return

        player: NetworkSlot = self.slot_info[packet.slot]
        print(f">> {player.name}: {packet.message}")


async def input_loop(client: ChatClient):
    while True:
        msg: str = await aioconsole.ainput()
        await client.send(packets.Say(text=msg))


async def main():
    async with ChatClient(port=12345) as client:
        input_loop_task = asyncio.create_task(input_loop(client))
        await client.wait_closed(input_loop_task)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

What's happening step by step

• Custom __init__: The ChatClient adds a slot_info dictionary to store player information. When overriding __init__, you must call super().__init__(port) to initialize the parent Client class.

• on_ready: Unlike the Advanced Client which authenticates in on_room_info, this example authenticates immediately in on_ready. Both approaches work — the difference is that on_room_info gives you access to the server's version.

• on_connected: Saves the slot_info dictionary from the server response. This maps slot numbers (integers) to NetworkSlot objects containing the player's name, game, etc. We need this later to display who sent a chat message.

• on_print_json: The server sends many types of PrintJSON messages (chat, item sends, player joins, hints, etc.). This callback checks packet.type and only processes CHAT messages, ignoring everything else. It then looks up the sender's name using self.slot_info[packet.slot].

• input_loop: Runs as a concurrent task alongside the client. aioconsole.ainput() waits for the user to type a line, then packets.Say(text=msg) sends it as a chat message to all players.

• except KeyboardInterrupt: Catches Ctrl+C so the program exits cleanly without printing a traceback.