The Quest Continues: Porting the Word Game With AsyncSSH

Last week, we reimplemented our 20-questions variant, and it received quite a bit of attention. So far we built a web experience, as well as a command line interface for the game. However, unlike the web version, the command line interface is strictly local. There’s no way to share the game with a fellow friend unless we walk them through the setup. Coincidentally, a former coworker recently told me about hosting applications over secure shell protocol in Golang. That makes me wonder, how easily can we replicate it in Python? Leveling Up LLM Game Development with DSPy A cute illustration from Copilot on the topic. The Spark of an idea The secure shell protocol (SSH) is usually what we use to log into another machine for work. Yet, there are a lot of interesting applications hosted over SSH. I remember there was one that would play back a Star Wars movie in ASCII art upon connecting successfully. Late last year, my coworker told me about a coffee shop selling freshly brewed coffee over SSH. We both found that idea intriguing. He suspected the application was made with wish, a Golang library. That prompted me to see if there is an equivalent for Python, but I only found Paramiko at the time. After a quick glance at the documentation, I decided it was too much work and moved on. It does seem like it is capable of achieving the goal — serving a CLI application over SSH, though. Yes, the em-dash is intentional. Photo by Vitaly Gariev on Unsplash Keeping in touch as usual, and he would bring up the wish project from time-to-time. Then fast-forward to last week, after I was done with the reimplementation of our little word game, I stumbled upon another library called AsyncSSH. Like Paramiko, it can be used to create an SSH server as well. The relatively more extensive examples also seemed helpful. After some exchanges with Gemini, I received a generated proof-of-concept code that looked workable. After completing and publishing the article on the game, I finally had time to properly explore the topic. Read on to join the ride together in building an interactive application served over SSH. Finding the Right Tool: Paramiko vs AsyncSSH Like we briefly mentioned previously, Paramiko and AsyncSSH are libraries we can use to build SSH Server applications (they are also excellent as SSH client libraries). Though the former garnered more stars on GitHub, the amount of information on how to build an SSH Server application is surprisingly sparse. From what I read, it also required quite extensive setup. The discovery of AsyncSSH was completely unintentional. It was buried deep in the search result when I was looking for a cohesive tutorial on Paramiko. While the design of the documentation site looked as if it was built in Python 2.0 era, the extensive collection of examples is extremely helpful. I particularly liked the simplicity offered by AsyncSSH in comparison. Photo by Agence Olloweb on Unsplash Our goal for this week is a fairly simple application sporting a simple REPL-like experience. The key of this exercise is to learn, hence efficiency isn’t a deciding factor. Therefore, the simplicity and developer friendly documentation make the library a sensible choice for our project this week. Two Paths and a Relearning Photo by Ross Sneddon on Unsplash The quick proof-of-concept code I came out with the help of Gemini, by referring to the documentation, consists of 3 components. Firstly, we have the actual SSH Server class, subclassing asyncssh.SSHServer. import asyncssh class MySSHServer(asyncssh.SSHServer): def begin_auth(self, username): logger.info("logging in user:%s", username) return False def session_requested(self): return MyREPLSession() This class is where we define all the authentication rules. For the purpose of this article, we are skipping all of them to maintain simplicity. Thus, we are returning False in begin_auth to indicate no authentication is needed. Do refer to the documentation for information on the topic before publishing applications in production. The MyREPLSession is a class inheriting asyncssh.SSHServerSession, which defines the actual application logic we intend to serve over the server. The skeleton is shown as below: class MyREPLSession(asyncssh.SSHServerSession): _chan = None _buffer = "" def connection_made(self, chan: asyncssh.SSHLineEditorChannel) -> None: self._chan = chan def pty_requested( self, term_type: str, term_size: tuple[int, int, int, int], term_modes: dict[int, int], ) -> bool: logger.info("pty received") return True def shell_requested(self) -> bool: logger.info("shell received") return True def data_received(self, data: str, datatype: int | None) -> None: # TODO The actual application pass def eof_received(self): assert isinstance(self._chan, asyn

May 4, 2025 - 21:06

The Quest Continues: Porting the Word Game With AsyncSSH

Last week, we reimplemented our 20-questions variant, and it received quite a bit of attention. So far we built a web experience, as well as a command line interface for the game. However, unlike the web version, the command line interface is strictly local. There’s no way to share the game with a fellow friend unless we walk them through the setup. Coincidentally, a former coworker recently told me about hosting applications over secure shell protocol in Golang. That makes me wonder, how easily can we replicate it in Python?

Leveling Up LLM Game Development with DSPy

A cute illustration from Copilot on the topic.

The Spark of an idea

The secure shell protocol (SSH) is usually what we use to log into another machine for work. Yet, there are a lot of interesting applications hosted over SSH. I remember there was one that would play back a Star Wars movie in ASCII art upon connecting successfully. Late last year, my coworker told me about a coffee shop selling freshly brewed coffee over SSH. We both found that idea intriguing.

He suspected the application was made with wish, a Golang library. That prompted me to see if there is an equivalent for Python, but I only found Paramiko at the time. After a quick glance at the documentation, I decided it was too much work and moved on. It does seem like it is capable of achieving the goal — serving a CLI application over SSH, though.

Yes, the em-dash is intentional.

Photo by Vitaly Gariev on Unsplash

Keeping in touch as usual, and he would bring up the wish project from time-to-time. Then fast-forward to last week, after I was done with the reimplementation of our little word game, I stumbled upon another library called AsyncSSH. Like Paramiko, it can be used to create an SSH server as well. The relatively more extensive examples also seemed helpful.

After some exchanges with Gemini, I received a generated proof-of-concept code that looked workable. After completing and publishing the article on the game, I finally had time to properly explore the topic. Read on to join the ride together in building an interactive application served over SSH.

Finding the Right Tool: Paramiko vs AsyncSSH

Like we briefly mentioned previously, Paramiko and AsyncSSH are libraries we can use to build SSH Server applications (they are also excellent as SSH client libraries). Though the former garnered more stars on GitHub, the amount of information on how to build an SSH Server application is surprisingly sparse. From what I read, it also required quite extensive setup.

The discovery of AsyncSSH was completely unintentional. It was buried deep in the search result when I was looking for a cohesive tutorial on Paramiko. While the design of the documentation site looked as if it was built in Python 2.0 era, the extensive collection of examples is extremely helpful. I particularly liked the simplicity offered by AsyncSSH in comparison.

Photo by Agence Olloweb on Unsplash

Our goal for this week is a fairly simple application sporting a simple REPL-like experience. The key of this exercise is to learn, hence efficiency isn’t a deciding factor. Therefore, the simplicity and developer friendly documentation make the library a sensible choice for our project this week.

Two Paths and a Relearning

Photo by Ross Sneddon on Unsplash

The quick proof-of-concept code I came out with the help of Gemini, by referring to the documentation, consists of 3 components. Firstly, we have the actual SSH Server class, subclassing asyncssh.SSHServer.

import asyncssh

class MySSHServer(asyncssh.SSHServer):
    def begin_auth(self, username):
        logger.info("logging in user:%s", username)
        return False

    def session_requested(self):
        return MyREPLSession()

This class is where we define all the authentication rules. For the purpose of this article, we are skipping all of them to maintain simplicity. Thus, we are returning False in begin_auth to indicate no authentication is needed. Do refer to the documentation for information on the topic before publishing applications in production.

The MyREPLSession is a class inheriting asyncssh.SSHServerSession, which defines the actual application logic we intend to serve over the server. The skeleton is shown as below:

class MyREPLSession(asyncssh.SSHServerSession):
    _chan = None
    _buffer = ""

    def connection_made(self, chan: asyncssh.SSHLineEditorChannel) -> None:
        self._chan = chan

    def pty_requested(
        self,
        term_type: str,
        term_size: tuple[int, int, int, int],
        term_modes: dict[int, int],
    ) -> bool:
        logger.info("pty received")
        return True

    def shell_requested(self) -> bool:
        logger.info("shell received")
        return True

    def data_received(self, data: str, datatype: int | None) -> None:
        # TODO The actual application
        pass

    def eof_received(self):
        assert isinstance(self._chan, asyncssh.SSHLineEditorChannel)

        self._chan.exit(0)
        return False

    def signal_received(self, signal):
        assert isinstance(self._chan, asyncssh.SSHLineEditorChannel)

        self._chan.exit(0)

    def break_received(self, signal):
        assert isinstance(self._chan, asyncssh.SSHLineEditorChannel)

        self._chan.exit(0)

There are a couple of key methods we need to implement. The first is the connection_made, in our setup we are expecting a asyncssh.SSHLineEditorChannel object. All the input and output through SSH would be handled by the channel object, hence it is crucial to keep a reference of it. Methods ending with the _received suffix are event handlers, and in our toy project, we simply close the channel when no more input is sent (upon EOF or on terminate/kill signals).

We will discuss data_received event handler later.

Lastly, we need to tie everything together through asyncssh.create_server. As implied by the library name, the function returns a coroutine object, and hence we would need to properly set up through AsyncIO. Feel free to revisit our previous article on how to write an asynchronous program. But for this example, we are adapting the example shown in the documentation.

AsyncIO Task Management: A Hands-On Scheduler Project

import asyncio

async def main() -> None:
    try:
        # Ensure ssh-host key exists before running
        await asyncssh.create_server(
            MySSHServer,
            "0.0.0.0",
            8022,
            server_host_keys=["./ssh-host"],
            line_editor=True
        )
        print("SSH server listening on port 8022...")
    except Exception as exc:
        print(f"Error starting server: {exc}")

if __name__ == " __main__":
    loop = asyncio.new_event_loop()

    loop.run_until_complete(main())
    loop.run_forever()

We previously used asyncio.run to run the main coroutine, and set up a ShutdownHandler working with an exit event object to end a program. On the other hand, this approach was likely written before asyncio.run was introduced, with the manual management of event loop being the telling sign. The run_forever() here also eliminates the hack where people put an infinite loop after asyncssh.create_server.

Essentially, the code snippet creates the server, where the server behaviour is defined by the MySSHServer, and we set line_editor=True to be explicit (as we are expecting user interaction in our session). Remember to create an SSH key for the application.

$ ssh-keygen -q -N "" -t ed25519 -f ./ssh-host

As mentioned earlier, we are building a very simple REPL-like application. Just as a refresher, REPL stands for read-evaluate-print-loop, which aptly fits our case here. We read user input, get DSPy to process it with the LLM in background, print the generated response, rinse-and-repeat. The skeleton of the flow is shown as below:

def main() -> None:
    game_setup_and_welcome_speech()

    while True:
        match input("> "):
            case command if command == "quit":
                break
            case attempt:
                process_and_print_reply(attempt)

Now we can populate our data_received handler. This method is called whenever our user is sending input through the SSH session. Unfortunately, we cannot assume the data is always ending with a newline (user pressing enter). Sometimes, the handler may have accumulated a couple of questions and send them in one go. The function may also get called while the user hasn’t done typing. For this, we would have to handle both cases, as shown below:

class MyREPLSession(asyncssh.SSHServerSession):
    _buffer = ""

    def data_received(self, data: str, datatype: int | None) -> None:
        assert isinstance(self._chan, asyncssh.SSHLineEditorChannel)

        self._chan.write(f"Welcome to my SSH server, {username}!\n\n")
        self._buffer += data

        lines = self._buffer.split("\n")
        for line in lines[:-1]:
            match line:
                case command if command == "quit":
                    self._chan.exit(0)
                    break

                case attempt if attempt:
                    process_and_print_reply(attempt)

        self._buffer = lines[-1]

Remember we talked about handling user input and output through the SSHLineEditingChannel object we received in connection_made? Output of our application is done through its write() method, just like how we output to stdout (to be displayed on the terminal emulator) through print() usually.

Let’s test the application.

A seemingly stalled game session

While the application worked, the experience wasn’t ideal. My modest GPU took quite a bit of time to generate a response for every user input, and somehow the cursor stalled and wouldn’t move to another line whenever the user pressed enter. It somehow gave an impression of the application just died after every enter.

Photo by Josh Redd on Unsplash

I was ready to move on, and start preparing to write, until I came across another example that fits the use-case. Now we have a code that looked a lot like the original skeleton.

async def handle_client(process: asyncssh.SSHServerProcess) -> None:
    username = process.get_extra_info("username")
    process.stdout.write(f"Welcome to my SSH server, {username}!\n\n")

    game_setup_and_welcome_speech()

    while True:
        match await process.stdin.readline():
            case command if command == "quit":
                break

            case attempt:
                process_and_print_reply(attempt)

    process.exit(0)

Previously we returned output through the SSHLineEditingChannel object, here we are doing it through the SSHServerProcess object. On the other hand, instead of input(), we read it through the readline() method in the stdin (an SSHReader object) property.

Then, we need to remove the session_requested method from MySSHServer, and define the process_factory argument in asyncssh.create_server to point to our new coroutine.

await asyncssh.create_server(
    MySSHServer,
    "0.0.0.0",
    8022,
    server_host_keys=["./ssh-host"],
    line_editor=True,
    process_factory=handle_client,
)

Re-run the application again, and to my surprise, it worked almost exactly like the original CLI version. For our project, the second approach definitely worked better. Nonetheless, the first attempt was also crucial, as it gives an idea on how input is being sent in a SSH session.

The choice of naming the property holding a SSHReader object as stdin, and the fact it reads input through a method named readline piqued my curiosity. I knew we could reach sys.stdin to read content redirected from the previous command (via the pipe | operator), or file content via the < operator (in Unix-like environments). That said, I didn’t know reading user input interactively in a running application is done through stdin as well.

My first experience writing a production CLI application started with PHP (very strange, yes). That was my first time learning about the relationship between the operators and stdin, stdout, and stderr through my supervisor. The scripts didn’t really need to collect input interactively, and given the similarity of PHP and C, I couldn’t find a scanf equivalent so I assumed PHP doesn’t have such feature.

I was wrong.

Photo by Mr. Bochelly on Unsplash

After some exchanges with Gemini, I realized I could use readline() in PHP for the task. Which itself is similar to scanf and input in Python. Alternatively, I could also use fgets(STDIN) to do the job, just that I need to figure out a way to print a prompt prior to the input collection. On the other hand, if I redirect some content into a Python application, if I call input(), it would read the content in stdin, until it encounters a newline character. Essentially, it roughly equates

import sys

sys.stdin.readline().strip()

Back to the topic, the toy example skipped many other aspects besides the authentication work we mentioned earlier. Unfortunately, due to the sheer scope and complexity, it is almost impossible for the developer to cover all use cases in the examples. Fortunately, the API documentation is very well-written, and I assume it could be useful for people who know more about the protocol.

We are lucky to live in an era where a LLM chatbot like Gemini is readily available, to provide help in navigating the documentation. Besides that, it also serves as a valuable companion to help gathering complementary information, like how I relearned stdin. Even so, critical thinking is still crucial while evaluating the response to avoid falling victim to AI hallucination.

Bridging Game and Server

Photo by Phil Hearing on Unsplash

We finally assembled all the pieces together and ported the game to be served over SSH. The concrete implementation can be found at the GitHub repository for those who are interested. Both the CLI version and SSH version (the handle_client variant) are very similar, as shown below

async def handle_client(process: asyncssh.SSHServerProcess) -> None:
    username = process.get_extra_info("username")
    process.stdout.write(f"Welcome to my SSH server, {username}!\n\n") # vs print()

    game_setup_and_welcome_speech()
    while True:
        match await process.stdin.readline(): # vs input()
            case command if command == "quit":
                break
            case command if command == "end" and answer is not None:
                answer = None
            case attempt:
                process_and_print_reply(attempt)

    process.exit(0)

The major difference between the two, is the use of process.stdout.write() to replace print(), and process.stdin.readline() to replace input (though a new coroutine called prompt_input is written to compose a process.stdout.write() to display prompt text).

async def prompt_input(process: asyncssh.SSHServerProcess, prompt: str) -> str:
    process.stdout.write(prompt)
    return await process.stdin.readline()

The REPL-like experience makes the word game a very synchronous experience. A response is only generated upon user submission of input. Therefore the await is necessary while expecting user input is needed to ensure order. Other than that, the asynchronous nature of AsyncSSH does not affect the game’s logic much. As each game session is independent, each time a new user connects, the library will run handle_client separately for each of them.

Once the log message shows “SSH server is listening …” after executing ssh.py, it is ready to be connected to with the following command.

$ ssh -p 8022 localhost

Considering we didn’t define any authentication methods in MySSHServer (which is definitely a bad practice), it should display a welcome message and the game is started right away.

In this article, we explored two different strategies to host our little word game over SSH. The two approaches have their own pros and cons, but it does show how AsyncSSH makes the development experience quite intuitive. Even though I do not have much experience doing this in Golang with wish (the documentation situation is also quite bad there), I enjoy the time spent doing this with AsyncSSH.

The Game Goes Live

Photo by Alessandro Erbetta on Unsplash

The point of the exercise is mainly on learning and discovery. If our game were developed with a text-based UI through Textual, or even Typer, some extra work is needed to ensure I/O is redirected properly.

The experience of porting the library to SSH is a series of happy coincidences. It started with a spark from a discussion with my former coworker, and it led to a series of discoveries. This article is a culmination of the whole journey, sharing what I learned throughout. It turns out building an application to be served over SSH just needed a change in how I/O is done. Gemini also helped immensely throughout, deepening my understanding.

We went through the whole process by configuring a server by subclassing a asyncssh.SSHServer, exploring two different ways to serve applications through process_factory and inheriting asyncssh.SSHServerSession, then getting it running via asyncssh.create_server.

Hopefully this article sparks an interest and we see more interesting applications hosted over SSH.

This article was written with assistance from Gemini, a large language model by Google. Gemini provided editorial help, fixing language errors and checking for article flow. However, the code and voice are entirely my own. If you’d like to collaborate or discuss job opportunities, please connect with me on LinkedIn or via Medium.