bottom is a small no-dependency async library for running simple or complex IRC clients
and requires python 3.12+
It's easy to get started with built-in support for common commands, and extensible enough to support any capabilities, including custom encryption, local events, bridging, replication, custom- and multi- syntax compatibility, and more.
pip install bottom
The user guide and API reference are available here including examples for regex based routing of privmsg, custom encryption, and a full list of rfc2812 commands that are supported by default.
The following example creates a client that will:
- connect, identify itself, wait for MOTD, and join a channel
- respond to
PINGautomatically - respond to any
PRIVMSGsent directly to it, or in a channel
import asyncio
import bottom
host = "irc.libera.chat"
port = 6697
ssl = True
NICK = "bottom-bot"
CHANNEL = "#bottom-dev"
bot = bottom.Client(host=host, port=port, ssl=ssl)
@bot.on('CLIENT_CONNECT')
async def connect(**kwargs):
await bot.send('nick', nick=NICK)
await bot.send('user', user=NICK,
realname='https://github.com/numberoverzero/bottom')
# Don't try to join channels until we're past the MOTD
await bottom.wait_for(bot, ["RPL_ENDOFMOTD", "ERR_NOMOTD"])
await bot.send('join', channel=CHANNEL)
@bot.on('PING')
async def keepalive(message: str, **kwargs):
await bot.send('pong', message=message)
@bot.on('PRIVMSG')
async def message(nick: str, target: str, message: str, **kwargs):
if nick == NICK:
return # bot sent this message, ignore
if target == NICK:
target = nick # direct message, respond directly
# else: respond in channel
await bot.send("privmsg", target=target, message=f"echo: {message}")
async def main():
await bot.connect()
try:
# serve until the connection drops...
await bot.wait("client_disconnect")
print("\ndisconnected by remote")
except asyncio.CancelledError:
# ...or we hit ctrl+c
await bot.disconnect()
print("\ndisconnected after ctrl+c")
if __name__ == "__main__":
asyncio.run(main())The public API that you'll typically interact with is small: the Client class and possibly register_pattern.
It is built around sending commands with send(cmd, **kw) (or send_message(msg) for raw IRC lines) and processing
events with @on(event) and wait(event).
If you need to customize serialization, message handling, or signal processing, those are all available with examples in the Extensions documentation.
class Client:
# true when the underlying connection is closed or closing
is_closing() -> bool:
# connects to the given host, port, and optionally over ssl.
async connect() -> None
# start disconnecting if connected. safe to call multiple times.
async disconnect() -> None
# send a known rfc2812 command, formatting kwargs for you
async send(command: str, **kwargs) -> None
# decorate a function (sync or async) to handle an event.
# these can be rfc2812 events (privmsg, ping, notice) or built-in
# events (client_connect, client_disconnect) or your own signals
@on(event: str)(async handler)
# manually trigger an event to be processed by any registered handlers
# for example, to simulate receiving a message:
# my_client.trigger("privmsg", nick=...)
# or send a local-only message to another part of your system:
# trigger("backup-local", backend="s3", session=...)
trigger(event: str, **kwargs) -> asyncio.Task
# wait for an event to be triggered.
async wait(event: str) -> dict
# send raw IRC line. bypasses rfc2812 parsing and validation,
# so you can support custom IRC messages or extensions, like SASL.
async send_message(message: str) -> None
# functions that handle the inbound raw IRC lines.
# by default, Client includes an rfc2812 handler that triggers
# events caught by @Client.on
message_handlers: list[ClientMessageHandler]
# register a new pattern for outbound serialization eg.
# register_pattern("MYCMD", "MYCMD {nick} {target}")
# client.send("MYCMD", nick="n0", target="remote.net")
def register_pattern(command: str, template: str)
# wait for the client to emit one or more events. when mode is "first"
# this returns the events that finished first (more than one event can be triggered
# in a single loop step) and cancels the rest. when mode is "all" this waits
# for all events to trigger.
async def wait_for(client, events: list[str], mode: "first"|"all") -> list[dict]
# helper classes to customize [de]serializing `dict <--> IRC line`
class CommandSerializer:
# register a new serialization pattern. these are sorted and looked up
# when trying to match input params against a valid command format.
# (some commands have multiple formats, like "TOPIC")
def register(command: str, template: str) -> SerializedTemplate
# format a dict of params into the best match template for a given command.
# searches all registered templates for that command from most args -> least args
# until a match is found and applied.
def serialize(command: str, params: dict) -> str
class SerializedTemplate:
# like string.format() but less overhead. applies custom formatters.
def format(params: dict) -> str
# returns an optimized version of string.format() that can use custom formatters
# eg. "{foo:myfunc} said {bar:join_commas}"
@classmethod
def parse(template: str, formatters: dict[str, Callable]) -> SerializedTemplate
# type hints for message handlers
type NextMessageHandler[T: Client] = Callable[[bytes], T, Coroutine[Any, Any, Any]]
type ClientMessageHandler[T: Client] = Callable[[NextMessageHandler[T], T, bytes], Coroutine[Any, Any, Any]]