Examples

If you’ve built something with matrix-nio and want to support the project, add a shield!

[![Built with matrix-nio](https://img.shields.io/badge/built%20with-matrix--nio-brightgreen)](https://github.com/poljar/matrix-nio)

To start making a chat bot quickly, considering using nio-template.

Attention

For E2EE support, python-olm is needed, which requires the libolm C library (version 3.x). After libolm has been installed, the e2ee enabled version of nio can be installed using pip install "matrix-nio[e2e]".

Projects built with nio

• pantalaimon - e2ee-aware reverse proxy daemon

• nio-template - template for creating bots

• matrix-tweetalong-bot - Twitter-backed watchalongs

• weechat-matrix - lets Weechat communicate over Matrix

• hemppa - generic modular Matrix bot

• devops-bot - sysadmin Swiss army knife bot for team efficiency

• podbot - play podcasts sent from AntennaPod via Riot.im elsewhere

• delator - bot with logs and modular commands

• cody - REPL for your matrix chat rooms

• matrix-nio-send - predecessor of matrix-commander

• Mirage - feature-rich keyboard-operable chat client written in Qt/QML

• matrix-commander - CLI-oriented integratable client

• matrix-archive - creates a YAML log of all room messages and media

• matrix-eno-bot - admin and personal assistence bot

• infinigpt-matrix - OpenAI GPT chatbot with infinite personalities

• ollamarama-matrix - AI chatbot with infinite personalities, using local LLMs via Ollama

• matrix-discord-bridge

• Matrix Nio - HACS - A HACS Integration to replace the core matrix Home Assistant integration

• Simple-Matrix-Bot-Lib

• pushmatrix

• matrix-webhook

• matrix-asgi

• opsdroid

• niobot - An extensive framework for building powerful Matrix bots with ease

• matrix-reminder-bot - A bot to remind you about stuff

• bartleby - A LLM chatbot and writing assistant using HuggingFace

Are we missing a project? Submit a pull request and we’ll get you added! Just edit doc/built-with-nio.rst

A basic client

A basic client requires a few things before you start:

• nio is installed

• a Matrix homeserver URL (probably “https://matrix.example.org”)

• a username and password for an account on that homeserver

• a room ID for a room on that homeserver. In Riot, this is found in the Room’s settings page under “Advanced”

By far the easiest way to use nio is using the asyncio layer, unless you have special restrictions that disallow the use of asyncio.

All examples require Python 3.5+ for the async / await syntax.

import asyncio

from nio import AsyncClient, MatrixRoom, RoomMessageText


async def message_callback(room: MatrixRoom, event: RoomMessageText) -> None:
    print(
        f"Message received in room {room.display_name}\n"
        f"{room.user_name(event.sender)} | {event.body}"
    )


async def main() -> None:
    client = AsyncClient("https://matrix.example.org", "@alice:example.org")
    client.add_event_callback(message_callback, RoomMessageText)

    print(await client.login("my-secret-password"))
    # "Logged in as @alice:example.org device id: RANDOMDID"

    # If you made a new room and haven't joined as that user, you can use
    # await client.join("your-room-id")

    await client.room_send(
        # Watch out! If you join an old room you'll see lots of old messages
        room_id="!my-fave-room:example.org",
        message_type="m.room.message",
        content={"msgtype": "m.text", "body": "Hello world!"},
    )
    await client.sync_forever(timeout=30000)  # milliseconds


asyncio.run(main())

Log in using a stored access_token

Using access tokens requires that when you first log in you save a few values to use later. In this example, we’re going to write them to disk as a JSON object, but you could also store them in a database, print them out and post them up on the wall beside your desk, text them to your sister in law, or anything else that allows you access to the values at a later date.

We’ve tried to keep this example small enough that it’s just enough to work; once you start writing your own programs with nio you may want to clean things up a bit.

This example requires that the user running it has write permissions to the folder they’re in. If you copied this repo to your computer, you probably have write permissions. Now run the program restore_login.py twice. First time around it will ask you for credentials like homeserver and password. On the second run, the program will log in for you automatically and it will send a “Hello World” message to the room you specify.

#!/usr/bin/env python3

import asyncio
import getpass
import json
import os
import sys

import aiofiles

from nio import AsyncClient, LoginResponse

CONFIG_FILE = "credentials.json"

# Check out main() below to see how it's done.


def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
    """Writes the required login details to disk so we can log in later without
    using a password.

    Arguments:
        resp {LoginResponse} -- the successful client login response.
        homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
    """
    # open the config file in write-mode
    with open(CONFIG_FILE, "w") as f:
        # write the login details to disk
        json.dump(
            {
                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
                "user_id": resp.user_id,  # e.g. "@user:example.org"
                "device_id": resp.device_id,  # device ID, 10 uppercase letters
                "access_token": resp.access_token,  # cryptogr. access token
            },
            f,
        )


async def main() -> None:
    # If there are no previously-saved credentials, we'll use the password
    if not os.path.exists(CONFIG_FILE):
        print(
            "First time use. Did not find credential file. Asking for "
            "homeserver, user, and password to create credential file."
        )
        homeserver = "https://matrix.example.org"
        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")

        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
            homeserver = "https://" + homeserver

        user_id = "@user:example.org"
        user_id = input(f"Enter your full user ID: [{user_id}] ")

        device_name = "matrix-nio"
        device_name = input(f"Choose a name for this device: [{device_name}] ")

        client = AsyncClient(homeserver, user_id)
        pw = getpass.getpass()

        resp = await client.login(pw, device_name=device_name)

        # check that we logged in successfully
        if isinstance(resp, LoginResponse):
            write_details_to_disk(resp, homeserver)
        else:
            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
            print(f"Failed to log in: {resp}")
            sys.exit(1)

        print(
            "Logged in using a password. Credentials were stored.",
            "Try running the script again to login with credentials.",
        )

    # Otherwise the config file exists, so we'll use the stored credentials
    else:
        # open the file in read-only mode
        async with aiofiles.open(CONFIG_FILE, "r") as f:
            contents = await f.read()
        config = json.loads(contents)
        client = AsyncClient(config["homeserver"])

        client.access_token = config["access_token"]
        client.user_id = config["user_id"]
        client.device_id = config["device_id"]

        # Now we can send messages as the user
        room_id = "!myfavouriteroomid:example.org"
        room_id = input(f"Enter room id for test message: [{room_id}] ")

        await client.room_send(
            room_id,
            message_type="m.room.message",
            content={"msgtype": "m.text", "body": "Hello world!"},
        )
        print("Logged in using stored credentials. Sent a test message.")

    # Either way we're logged in here, too
    await client.close()


asyncio.run(main())

Sending an image

Now that you have sent a first “Hello World” text message, how about going one step further and sending an image, like a photo from your last vacation. Run the send_image.py program and provide a filename to the photo. Voila, you have just sent your first image!

#!/usr/bin/env python3

import asyncio
import getpass
import json
import os
import sys

import aiofiles.os
import magic
from PIL import Image

from nio import AsyncClient, LoginResponse, UploadResponse

CONFIG_FILE = "credentials.json"

# Check out main() below to see how it's done.


def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
    """Writes the required login details to disk so we can log in later without
    using a password.

    Arguments:
        resp {LoginResponse} -- the successful client login response.
        homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
    """
    # open the config file in write-mode
    with open(CONFIG_FILE, "w") as f:
        # write the login details to disk
        json.dump(
            {
                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
                "user_id": resp.user_id,  # e.g. "@user:example.org"
                "device_id": resp.device_id,  # device ID, 10 uppercase letters
                "access_token": resp.access_token,  # cryptogr. access token
            },
            f,
        )


async def send_image(client, room_id, image):
    """Send image to room.

    Arguments:
    ---------
    client : Client
    room_id : str
    image : str, file name of image

    This is a working example for a JPG image.
        "content": {
            "body": "someimage.jpg",
            "info": {
                "size": 5420,
                "mimetype": "image/jpeg",
                "thumbnail_info": {
                    "w": 100,
                    "h": 100,
                    "mimetype": "image/jpeg",
                    "size": 2106
                },
                "w": 100,
                "h": 100,
                "thumbnail_url": "mxc://example.com/SomeStrangeThumbnailUriKey"
            },
            "msgtype": "m.image",
            "url": "mxc://example.com/SomeStrangeUriKey"
        }

    """
    mime_type = magic.from_file(image, mime=True)  # e.g. "image/jpeg"
    if not mime_type.startswith("image/"):
        print("Drop message because file does not have an image mime type.")
        return

    im = Image.open(image)
    (width, height) = im.size  # im.size returns (width,height) tuple

    # first do an upload of image, then send URI of upload to room
    file_stat = await aiofiles.os.stat(image)
    async with aiofiles.open(image, "r+b") as f:
        resp, maybe_keys = await client.upload(
            f,
            content_type=mime_type,  # image/jpeg
            filename=os.path.basename(image),
            filesize=file_stat.st_size,
        )
    if isinstance(resp, UploadResponse):
        print("Image was uploaded successfully to server. ")
    else:
        print(f"Failed to upload image. Failure response: {resp}")

    content = {
        "body": os.path.basename(image),  # descriptive title
        "info": {
            "size": file_stat.st_size,
            "mimetype": mime_type,
            "thumbnail_info": None,  # TODO
            "w": width,  # width in pixel
            "h": height,  # height in pixel
            "thumbnail_url": None,  # TODO
        },
        "msgtype": "m.image",
        "url": resp.content_uri,
    }

    try:
        await client.room_send(room_id, message_type="m.room.message", content=content)
        print("Image was sent successfully")
    except Exception:
        print(f"Image send of file {image} failed.")


async def main() -> None:
    # If there are no previously-saved credentials, we'll use the password
    if not os.path.exists(CONFIG_FILE):
        print(
            "First time use. Did not find credential file. Asking for "
            "homeserver, user, and password to create credential file."
        )
        homeserver = "https://matrix.example.org"
        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")

        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
            homeserver = "https://" + homeserver

        user_id = "@user:example.org"
        user_id = input(f"Enter your full user ID: [{user_id}] ")

        device_name = "matrix-nio"
        device_name = input(f"Choose a name for this device: [{device_name}] ")

        client = AsyncClient(homeserver, user_id)
        pw = getpass.getpass()

        resp = await client.login(pw, device_name=device_name)

        # check that we logged in successfully
        if isinstance(resp, LoginResponse):
            write_details_to_disk(resp, homeserver)
        else:
            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
            print(f"Failed to log in: {resp}")
            sys.exit(1)

        print(
            "Logged in using a password. Credentials were stored.",
            "Try running the script again to login with credentials.",
        )

    # Otherwise the config file exists, so we'll use the stored credentials
    else:
        # open the file in read-only mode
        async with aiofiles.open(CONFIG_FILE, "r") as f:
            contents = await f.read()
        config = json.loads(contents)
        client = AsyncClient(config["homeserver"])

        client.access_token = config["access_token"]
        client.user_id = config["user_id"]
        client.device_id = config["device_id"]

        # Now we can send messages as the user
        room_id = "!myfavouriteroomid:example.org"
        room_id = input(f"Enter room id for image message: [{room_id}] ")

        image = "exampledir/samplephoto.jpg"
        image = input(f"Enter file name of image to send: [{image}] ")

        await send_image(client, room_id, image)
        print("Logged in using stored credentials. Sent a test message.")

    # Close the client connection after we are done with it.
    await client.close()


asyncio.run(main())

Manual encryption key verification

Below is a program that works through manual encryption of other users when you already know all of their device IDs. It’s a bit dense but provides a good example in terms of being pythonic and using nio’s design features purposefully. It is not designed to be a template that you can immediately extend to run your bot, it’s designed to be an example of how to use nio.

The overall structure is this: we subclass nio’s AsyncClient class and add in our own handlers for a few things, namely:

• automatically restoring login details from disk instead of creating new sessions each time we restart the process

• callback for printing out any message we receive to stdout

• callback for automatically joining any room @alice is invited to

• a method for trusting devices using a user ID and (optionally) their list of trusted device IDs

• a sample “hello world” encrypted message method

In main, we make an instance of that subclass, attempt to login, then create an asyncio coroutine to run later that will trust the devices and send the hello world message. We then create `asyncio Tasks <>`_ to run that coroutine as well as the sync_forever() coroutine that nio provides, which does most of the handling of required work for communicating with Matrix: it uploads keys, checks for new messages, executes callbacks when events occur that trigger those callbacks, etc. Main executes the result of those Tasks.

You’ll need two accounts, which we’ll call @alice:example.org and @bob:example.org. @alice will be your nio application and @bob will be your second user account. Before the script runs, make a new room with the @bob account, enable encryption and invite @alice. Note the room ID as you’ll need it for this script. You’ll also need all of @bob’s device IDs, which you can get from within Riot under the profile settings > Advanced section. They may be called “session IDs”. These are the device IDs that your program will trust, and getting them into nio is the manual part here. In another example we’ll document automatic emoji verification.

It may look long at first but much of the program is actually documentation explaining how it works. If you have questions about the example, please don’t hesitate to ask them on #nio:matrix.org.

If you are stuck, it may be useful to read this primer from Matrix.org on implementing end-to-end encryption: https://matrix.org/docs/guides/end-to-end-encryption-implementation-guide

To delete the store, or clear the trusted devices, simply remove “nio_store” in the working directory as well as “manual_encrypted_verify.json”. Then the example script will log in (with a new session ID) and generate new keys.

import asyncio
import json
import os
import sys
from typing import Optional

import aiofiles

from nio import (
    AsyncClient,
    ClientConfig,
    InviteEvent,
    LoginResponse,
    MatrixRoom,
    RoomMessageText,
    crypto,
    exceptions,
)

# This is a fully-documented example of how to do manual verification with nio,
# for when you already know the device IDs of the users you want to trust. If
# you want live verification using emojis, the process is more complicated and
# will be covered in another example.

# We're building on the restore_login example here to preserve device IDs and
# therefore preserve trust; if @bob trusts @alice's device ID ABC and @alice
# restarts this program, loading the same keys, @bob will preserve trust. If
# @alice logged in again @alice would have new keys and a device ID XYZ, and
# @bob wouldn't trust it.

# The store is where we want to place encryption details like our keys, trusted
# devices and blacklisted devices. Here we place it in the working directory,
# but if you deploy your program you might consider /var or /opt for storage
STORE_FOLDER = "nio_store/"

# This file is for restoring login details after closing the program, so you
# can preserve your device ID. If @alice logged in every time instead, @bob
# would have to re-verify. See the restoring login example for more into.
SESSION_DETAILS_FILE = "credentials.json"

# Only needed for this example, this is who @alice will securely
# communicate with. We need all the device IDs of this user so we can consider
# them "trusted". If an unknown device shows up (like @bob signs into their
# account on another device), this program will refuse to send a message in the
# room. Try it!
BOB_ID = "@bob:example.org"
BOB_DEVICE_IDS = [
    # You can find these in Riot under Settings > Security & Privacy.
    # They may also be called "session IDs". You'll want to add ALL of them here
    # for the one other user in your encrypted room
    "URDEVICEID",
]

# the ID of the room you want your bot to join and send commands in.
# This can be a direct message or room; Matrix treats them the same
ROOM_ID = "!myfavouriteroom:example.org"

ALICE_USER_ID = "@alice:example.org"
ALICE_HOMESERVER = "https://matrix.example.org"
ALICE_PASSWORD = "hunter2"


class CustomEncryptedClient(AsyncClient):
    def __init__(
        self,
        homeserver,
        user="",
        device_id="",
        store_path="",
        config=None,
        ssl=None,
        proxy=None,
    ):
        # Calling super.__init__ means we're running the __init__ method
        # defined in AsyncClient, which this class derives from. That does a
        # bunch of setup for us automatically
        super().__init__(
            homeserver,
            user=user,
            device_id=device_id,
            store_path=store_path,
            config=config,
            ssl=ssl,
            proxy=proxy,
        )

        # if the store location doesn't exist, we'll make it
        if store_path and not os.path.isdir(store_path):
            os.mkdir(store_path)

        # auto-join room invites
        self.add_event_callback(self.cb_autojoin_room, InviteEvent)

        # print all the messages we receive
        self.add_event_callback(self.cb_print_messages, RoomMessageText)

    async def login(self) -> None:
        """Log in either using the global variables or (if possible) using the
        session details file.

        NOTE: This method kinda sucks. Don't use these kinds of global
        variables in your program; it would be much better to pass them
        around instead. They are only used here to minimise the size of the
        example.
        """
        # Restore the previous session if we can
        # See the "restore_login.py" example if you're not sure how this works
        if os.path.exists(SESSION_DETAILS_FILE) and os.path.isfile(
            SESSION_DETAILS_FILE
        ):
            try:
                async with aiofiles.open(SESSION_DETAILS_FILE, "r") as f:
                    contents = await f.read()
                config = json.loads(contents)
                self.access_token = config["access_token"]
                self.user_id = config["user_id"]
                self.device_id = config["device_id"]

                # This loads our verified/blacklisted devices and our keys
                self.load_store()
                print(
                    f"Logged in using stored credentials: {self.user_id} on {self.device_id}"
                )

            except OSError as err:
                print(f"Couldn't load session from file. Logging in. Error: {err}")
            except json.JSONDecodeError:
                print("Couldn't read JSON file; overwriting")

        # We didn't restore a previous session, so we'll log in with a password
        if not self.user_id or not self.access_token or not self.device_id:
            # this calls the login method defined in AsyncClient from nio
            resp = await super().login(ALICE_PASSWORD)

            if isinstance(resp, LoginResponse):
                print("Logged in using a password; saving details to disk")
                self.__write_details_to_disk(resp)
            else:
                print(f"Failed to log in: {resp}")
                sys.exit(1)

    def trust_devices(self, user_id: str, device_list: Optional[str] = None) -> None:
        """Trusts the devices of a user.

        If no device_list is provided, all of the users devices are trusted. If
        one is provided, only the devices with IDs in that list are trusted.

        Arguments:
            user_id {str} -- the user ID whose devices should be trusted.

        Keyword Arguments:
            device_list {Optional[str]} -- The full list of device IDs to trust
                from that user (default: {None})
        """

        print(f"{user_id}'s device store: {self.device_store[user_id]}")

        # The device store contains a dictionary of device IDs and known
        # OlmDevices for all users that share a room with us, including us.

        # We can only run this after a first sync. We have to populate our
        # device store and that requires syncing with the server.
        for device_id, olm_device in self.device_store[user_id].items():
            if device_list and device_id not in device_list:
                # a list of trusted devices was provided, but this ID is not in
                # that list. That's an issue.
                print(
                    f"Not trusting {device_id} as it's not in {user_id}'s pre-approved list."
                )
                continue

            if user_id == self.user_id and device_id == self.device_id:
                # We cannot explicitly trust the device @alice is using
                continue

            self.verify_device(olm_device)
            print(f"Trusting {device_id} from user {user_id}")

    def cb_autojoin_room(self, room: MatrixRoom, event: InviteEvent):
        """Callback to automatically joins a Matrix room on invite.

        Arguments:
            room {MatrixRoom} -- Provided by nio
            event {InviteEvent} -- Provided by nio
        """
        self.join(room.room_id)
        room = self.rooms[ROOM_ID]
        print(f"Room {room.name} is encrypted: {room.encrypted}")

    async def cb_print_messages(self, room: MatrixRoom, event: RoomMessageText):
        """Callback to print all received messages to stdout.

        Arguments:
            room {MatrixRoom} -- Provided by nio
            event {RoomMessageText} -- Provided by nio
        """
        if event.decrypted:
            encrypted_symbol = "🛡 "
        else:
            encrypted_symbol = "⚠️ "
        print(
            f"{room.display_name} |{encrypted_symbol}| {room.user_name(event.sender)}: {event.body}"
        )

    async def send_hello_world(self):
        # Now we send an encrypted message that @bob can read, although it will
        # appear to be "unverified" when they see it, because @bob has not verified
        # the device @alice is sending from.
        # We'll leave that as an exercise for the reader.
        try:
            await self.room_send(
                room_id=ROOM_ID,
                message_type="m.room.message",
                content={
                    "msgtype": "m.text",
                    "body": "Hello, this message is encrypted",
                },
            )
        except exceptions.OlmUnverifiedDeviceError:
            print("These are all known devices:")
            device_store: crypto.DeviceStore = device_store  # noqa: F821
            [
                print(
                    f"\t{device.user_id}\t {device.device_id}\t {device.trust_state}\t  {device.display_name}"
                )
                for device in device_store
            ]
            sys.exit(1)

    @staticmethod
    def __write_details_to_disk(resp: LoginResponse) -> None:
        """Writes login details to disk so that we can restore our session later
        without logging in again and creating a new device ID.

        Arguments:
            resp {LoginResponse} -- the successful client login response.
        """
        with open(SESSION_DETAILS_FILE, "w") as f:
            json.dump(
                {
                    "access_token": resp.access_token,
                    "device_id": resp.device_id,
                    "user_id": resp.user_id,
                },
                f,
            )


async def run_client(client: CustomEncryptedClient) -> None:
    """A basic encrypted chat application using nio."""

    # This is our own custom login function that looks for a pre-existing config
    # file and, if it exists, logs in using those details. Otherwise it will log
    # in using a password.
    await client.login()

    # Here we create a coroutine that we can call in asyncio.gather later,
    # along with sync_forever and any other API-related coroutines you'd like
    # to do.
    async def after_first_sync():
        # We'll wait for the first firing of 'synced' before trusting devices.
        # client.synced is an asyncio event that fires any time nio syncs. This
        # code doesn't run in a loop, so it only fires once
        print("Awaiting sync")
        await client.synced.wait()

        # In practice, you want to have a list of previously-known device IDs
        # for each user you want to trust. Here, we require that list as a
        # global variable
        client.trust_devices(BOB_ID, BOB_DEVICE_IDS)

        # In this case, we'll trust _all_ of @alice's devices. NOTE that this
        # is a SUPER BAD IDEA in practice, but for the purpose of this example
        # it'll be easier, since you may end up creating lots of sessions for
        # @alice as you play with the script
        client.trust_devices(ALICE_USER_ID)

        await client.send_hello_world()

    # We're creating Tasks here so that you could potentially write other
    # Python coroutines to do other work, like checking an API or using another
    # library. All of these Tasks will be run concurrently.
    # For more details, check out https://docs.python.org/3/library/asyncio-task.html

    # ensure_future() is for Python 3.5 and 3.6 compatibility. For 3.7+, use
    # asyncio.create_task()
    after_first_sync_task = asyncio.ensure_future(after_first_sync())

    # We use full_state=True here to pull any room invites that occurred or
    # messages sent in rooms _before_ this program connected to the
    # Matrix server
    sync_forever_task = asyncio.ensure_future(
        client.sync_forever(30000, full_state=True)
    )

    await asyncio.gather(
        # The order here IS significant! You have to register the task to trust
        # devices FIRST since it awaits the first sync
        after_first_sync_task,
        sync_forever_task,
    )


async def main():
    # By setting `store_sync_tokens` to true, we'll save sync tokens to our
    # store every time we sync, thereby preventing reading old, previously read
    # events on each new sync.
    # For more info, check out https://matrix-nio.readthedocs.io/en/latest/nio.html#asyncclient
    config = ClientConfig(store_sync_tokens=True)
    client = CustomEncryptedClient(
        ALICE_HOMESERVER,
        ALICE_USER_ID,
        store_path=STORE_FOLDER,
        config=config,
        ssl=False,
        proxy="http://localhost:8080",
    )

    try:
        await run_client(client)
    except (asyncio.CancelledError, KeyboardInterrupt):
        await client.close()


# Run the main coroutine, which instantiates our custom subclass, trusts all the
# devices, and syncs forever (or until your press Ctrl+C)

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

Interactive encryption key verification

One way to interactively verify a device is via emojis. On popular Matrix clients you will find that devices are flagged as trusted or untrusted. If a device is untrusted you can verify to make it trusted. Most clients have a red symbol for untrusted and a green icon for trusted. One can select un untrusted device and initiate a verify by emoji action. How would that look like in code? How can you add that to your application? Next we present a simple application that showcases emoji verification. Note, the app only accepts emoji verification. So, you have to start it on the other client (e.g. Element). Initiating an emoji verification is similar in code, consider doing it as “homework” if you feel up to it. But for now, let’s have a look how emoji verification can be accepted and processed.

#!/usr/bin/env python3

"""verify_with_emoji.py A sample program to demo Emoji verification.

# Objectives:
- Showcase the emoji verification using matrix-nio SDK
- This sample program tries to show the key steps involved in performing
    an emoji verification.
- It does so only for incoming request, outgoing emoji verification request
    are similar but not shown in this sample program

# Prerequisites:
- You must have matrix-nio and components for end-to-end encryption installed
    See: https://github.com/poljar/matrix-nio
- You must have created a Matrix account already,
    and have username and password ready
- You must have already joined a Matrix room with someone, e.g. yourself
- This other party initiates an emoji verification with you
- You are using this sample program to accept this incoming emoji verification
    and follow the protocol to successfully verify the other party's device

# Use Cases:
- Apply similar code in your Matrix bot
- Apply similar code in your Matrix client
- Just to learn about Matrix and the matrix-nio SDK

# Running the Program:
- Change permissions to allow execution
    `chmod 755 ./verify_with_emoji.py`
- Optionally create a store directory, if not it will be done for you
    `mkdir ./store/`
- Run the program as-is, no changes needed
    `./verify_with_emoji.py`
- Run it as often as you like

# Sample Screen Output when Running Program:
$ ./verify_with_emoji.py
First time use. Did not find credential file. Asking for
homeserver, user, and password to create credential file.
Enter your homeserver URL: [https://matrix.example.org] matrix.example.org
Enter your full user ID: [@user:example.org] @user:example.org
Choose a name for this device: [matrix-nio] verify_with_emoji
Password:
Logged in using a password. Credentials were stored.
On next execution the stored login credentials will be used.
This program is ready and waiting for the other party to initiate an emoji
verification with us by selecting "Verify by Emoji" in their Matrix client.
[('⚓', 'Anchor'), ('☎️', 'Telephone'), ('😀', 'Smiley'), ('😀', 'Smiley'),
 ('☂️', 'Umbrella'), ('⚓', 'Anchor'), ('☎️', 'Telephone')]
Do the emojis match? (Y/N) y
Match! Device will be verified by accepting verification.
sas.we_started_it = False
sas.sas_accepted = True
sas.canceled = False
sas.timed_out = False
sas.verified = True
sas.verified_devices = ['DEVICEIDXY']
Emoji verification was successful.
Hit Control-C to stop the program or initiate another Emoji verification
from another device or room.

"""

import asyncio
import getpass
import json
import os
import sys
import traceback

import aiofiles

from nio import (
    AsyncClient,
    AsyncClientConfig,
    KeyVerificationCancel,
    KeyVerificationEvent,
    KeyVerificationKey,
    KeyVerificationMac,
    KeyVerificationStart,
    LocalProtocolError,
    LoginResponse,
    ToDeviceError,
)

# file to store credentials in case you want to run program multiple times
CONFIG_FILE = "credentials.json"  # login credentials JSON file
# directory to store persistent data for end-to-end encryption
STORE_PATH = "./store/"  # local directory


class Callbacks:
    """Class to pass client to callback methods."""

    def __init__(self, client):
        """Store AsyncClient."""
        self.client = client

    async def to_device_callback(self, event):  # noqa
        """Handle events sent to device."""
        try:
            client = self.client

            if isinstance(event, KeyVerificationStart):  # first step
                """first step: receive KeyVerificationStart
                KeyVerificationStart(
                    source={'content':
                            {'method': 'm.sas.v1',
                             'from_device': 'DEVICEIDXY',
                             'key_agreement_protocols':
                                ['curve25519-hkdf-sha256', 'curve25519'],
                             'hashes': ['sha256'],
                             'message_authentication_codes':
                                ['hkdf-hmac-sha256', 'hmac-sha256'],
                             'short_authentication_string':
                                ['decimal', 'emoji'],
                             'transaction_id': 'SomeTxId'
                             },
                            'type': 'm.key.verification.start',
                            'sender': '@user2:example.org'
                            },
                    sender='@user2:example.org',
                    transaction_id='SomeTxId',
                    from_device='DEVICEIDXY',
                    method='m.sas.v1',
                    key_agreement_protocols=[
                        'curve25519-hkdf-sha256', 'curve25519'],
                    hashes=['sha256'],
                    message_authentication_codes=[
                        'hkdf-hmac-sha256', 'hmac-sha256'],
                    short_authentication_string=['decimal', 'emoji'])
                """

                if "emoji" not in event.short_authentication_string:
                    print(
                        "Other device does not support emoji verification "
                        f"{event.short_authentication_string}."
                    )
                    return
                resp = await client.accept_key_verification(event.transaction_id)
                if isinstance(resp, ToDeviceError):
                    print(f"accept_key_verification failed with {resp}")

                sas = client.key_verifications[event.transaction_id]

                todevice_msg = sas.share_key()
                resp = await client.to_device(todevice_msg)
                if isinstance(resp, ToDeviceError):
                    print(f"to_device failed with {resp}")

            elif isinstance(event, KeyVerificationCancel):  # anytime
                """at any time: receive KeyVerificationCancel
                KeyVerificationCancel(source={
                    'content': {'code': 'm.mismatched_sas',
                                'reason': 'Mismatched authentication string',
                                'transaction_id': 'SomeTxId'},
                    'type': 'm.key.verification.cancel',
                    'sender': '@user2:example.org'},
                    sender='@user2:example.org',
                    transaction_id='SomeTxId',
                    code='m.mismatched_sas',
                    reason='Mismatched short authentication string')
                """

                # There is no need to issue a
                # client.cancel_key_verification(tx_id, reject=False)
                # here. The SAS flow is already cancelled.
                # We only need to inform the user.
                print(
                    f"Verification has been cancelled by {event.sender} "
                    f'for reason "{event.reason}".'
                )

            elif isinstance(event, KeyVerificationKey):  # second step
                """Second step is to receive KeyVerificationKey
                KeyVerificationKey(
                    source={'content': {
                            'key': 'SomeCryptoKey',
                            'transaction_id': 'SomeTxId'},
                        'type': 'm.key.verification.key',
                        'sender': '@user2:example.org'
                    },
                    sender='@user2:example.org',
                    transaction_id='SomeTxId',
                    key='SomeCryptoKey')
                """
                sas = client.key_verifications[event.transaction_id]

                print(f"{sas.get_emoji()}")

                yn = input("Do the emojis match? (Y/N) (C for Cancel) ")
                if yn.lower() == "y":
                    print(
                        "Match! The verification for this " "device will be accepted."
                    )
                    resp = await client.confirm_short_auth_string(event.transaction_id)
                    if isinstance(resp, ToDeviceError):
                        print(f"confirm_short_auth_string failed with {resp}")
                elif yn.lower() == "n":  # no, don't match, reject
                    print(
                        "No match! Device will NOT be verified "
                        "by rejecting verification."
                    )
                    resp = await client.cancel_key_verification(
                        event.transaction_id, reject=True
                    )
                    if isinstance(resp, ToDeviceError):
                        print(f"cancel_key_verification failed with {resp}")
                else:  # C or anything for cancel
                    print("Cancelled by user! Verification will be " "cancelled.")
                    resp = await client.cancel_key_verification(
                        event.transaction_id, reject=False
                    )
                    if isinstance(resp, ToDeviceError):
                        print(f"cancel_key_verification failed with {resp}")

            elif isinstance(event, KeyVerificationMac):  # third step
                """Third step is to receive KeyVerificationMac
                KeyVerificationMac(
                    source={'content': {
                        'mac': {'ed25519:DEVICEIDXY': 'SomeKey1',
                                'ed25519:SomeKey2': 'SomeKey3'},
                        'keys': 'SomeCryptoKey4',
                        'transaction_id': 'SomeTxId'},
                        'type': 'm.key.verification.mac',
                        'sender': '@user2:example.org'},
                    sender='@user2:example.org',
                    transaction_id='SomeTxId',
                    mac={'ed25519:DEVICEIDXY': 'SomeKey1',
                         'ed25519:SomeKey2': 'SomeKey3'},
                    keys='SomeCryptoKey4')
                """
                sas = client.key_verifications[event.transaction_id]
                try:
                    todevice_msg = sas.get_mac()
                except LocalProtocolError as e:
                    # e.g. it might have been cancelled by ourselves
                    print(
                        f"Cancelled or protocol error: Reason: {e}.\n"
                        f"Verification with {event.sender} not concluded. "
                        "Try again?"
                    )
                else:
                    resp = await client.to_device(todevice_msg)
                    if isinstance(resp, ToDeviceError):
                        print(f"to_device failed with {resp}")
                    print(
                        f"sas.we_started_it = {sas.we_started_it}\n"
                        f"sas.sas_accepted = {sas.sas_accepted}\n"
                        f"sas.canceled = {sas.canceled}\n"
                        f"sas.timed_out = {sas.timed_out}\n"
                        f"sas.verified = {sas.verified}\n"
                        f"sas.verified_devices = {sas.verified_devices}\n"
                    )
                    print(
                        "Emoji verification was successful!\n"
                        "Hit Control-C to stop the program or "
                        "initiate another Emoji verification from "
                        "another device or room."
                    )
            else:
                print(
                    f"Received unexpected event type {type(event)}. "
                    f"Event is {event}. Event will be ignored."
                )
        except BaseException:
            print(traceback.format_exc())


def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
    """Write the required login details to disk.

    It will allow following logins to be made without password.

    Arguments:
    ---------
        resp : LoginResponse - successful client login response
        homeserver : str - URL of homeserver, e.g. "https://matrix.example.org"

    """
    # open the config file in write-mode
    with open(CONFIG_FILE, "w") as f:
        # write the login details to disk
        json.dump(
            {
                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
                "user_id": resp.user_id,  # e.g. "@user:example.org"
                "device_id": resp.device_id,  # device ID, 10 uppercase letters
                "access_token": resp.access_token,  # cryptogr. access token
            },
            f,
        )


async def login() -> AsyncClient:
    """Handle login with or without stored credentials."""
    # Configuration options for the AsyncClient
    client_config = AsyncClientConfig(
        max_limit_exceeded=0,
        max_timeouts=0,
        store_sync_tokens=True,
        encryption_enabled=True,
    )

    # If there are no previously-saved credentials, we'll use the password
    if not os.path.exists(CONFIG_FILE):
        print(
            "First time use. Did not find credential file. Asking for "
            "homeserver, user, and password to create credential file."
        )
        homeserver = "https://matrix.example.org"
        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")

        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
            homeserver = "https://" + homeserver

        user_id = "@user:example.org"
        user_id = input(f"Enter your full user ID: [{user_id}] ")

        device_name = "matrix-nio"
        device_name = input(f"Choose a name for this device: [{device_name}] ")

        if not os.path.exists(STORE_PATH):
            os.makedirs(STORE_PATH)

        # Initialize the matrix client
        client = AsyncClient(
            homeserver,
            user_id,
            store_path=STORE_PATH,
            config=client_config,
        )
        pw = getpass.getpass()

        resp = await client.login(password=pw, device_name=device_name)

        # check that we logged in successfully
        if isinstance(resp, LoginResponse):
            write_details_to_disk(resp, homeserver)
        else:
            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
            print(f"Failed to log in: {resp}")
            sys.exit(1)

        print(
            "Logged in using a password. Credentials were stored. "
            "On next execution the stored login credentials will be used."
        )

    # Otherwise the config file exists, so we'll use the stored credentials
    else:
        # open the file in read-only mode
        async with aiofiles.open(CONFIG_FILE, "r") as f:
            contents = await f.read()
        config = json.loads(contents)
        # Initialize the matrix client based on credentials from file
        client = AsyncClient(
            config["homeserver"],
            config["user_id"],
            device_id=config["device_id"],
            store_path=STORE_PATH,
            config=client_config,
        )

        client.restore_login(
            user_id=config["user_id"],
            device_id=config["device_id"],
            access_token=config["access_token"],
        )
        print("Logged in using stored credentials.")

    return client


async def main() -> None:
    """Login and wait for and perform emoji verify."""
    client = await login()
    # Set up event callbacks
    callbacks = Callbacks(client)
    client.add_to_device_callback(callbacks.to_device_callback, (KeyVerificationEvent,))
    # Sync encryption keys with the server
    # Required for participating in encrypted rooms
    if client.should_upload_keys:
        await client.keys_upload()
    print(
        "This program is ready and waiting for the other party to initiate "
        'an emoji verification with us by selecting "Verify by Emoji" '
        "in their Matrix client."
    )
    await client.sync_forever(timeout=30000, full_state=True)


try:
    asyncio.run(main())
except Exception:
    print(traceback.format_exc())
    sys.exit(1)
except KeyboardInterrupt:
    print("Received keyboard interrupt.")
    sys.exit(0)

Further reading and exploration

In an external repo, not maintained by us, is a simple Matrix client that includes sending, receiving and verification. It gives an example of

• how to send text, images, audio, video, other text files

• listen to messages forever

• get just the newest unread messages

• get the last N messages

• perform emoji verification

• etc.

So, if you want more example code and want to explore further have a look at this external repo called matrix-commander. And of course, you should check out all the other projects built with matrix-nio. To do so, check out our built-with-marix-nio list.