tessl/pypi-irc
IRC (Internet Relay Chat) protocol library for Python
—
Pending
bot-framework.mddocs/
Bot Framework
High-level IRC bot framework with automatic reconnection, channel management, and common bot functionality. Provides SingleServerIRCBot base class for easy bot development with built-in channel state tracking and reconnection strategies.
Capabilities
SingleServerIRCBot
Main IRC bot class that extends SimpleIRCClient with bot-specific functionality including automatic reconnection, channel management, and CTCP handling.
class SingleServerIRCBot:
def __init__(self, server_list, nickname, realname,
_=None, recon=ExponentialBackoff(), **connect_params):
"""
Initialize IRC bot for single server connection.
Parameters:
- server_list: list, server specifications (ServerSpec objects or dicts)
- nickname: bot nickname
- realname: bot real name
- _: unused parameter (default: None)
- recon: ReconnectStrategy, reconnection strategy (default: ExponentialBackoff())
- **connect_params: additional connection parameters
"""
@property
def channels(self) -> dict:
"""Dictionary mapping channel names to Channel objects."""
@property
def servers(self) -> list:
"""List of server specifications."""
@property
def recon(self) -> ReconnectStrategy:
"""Current reconnection strategy."""
def start(self):
"""
Start bot operation and event processing.
Connects to server and begins event loop.
"""
def die(self, msg: str = "Bye, cruel world!"):
"""
Terminate bot permanently.
Parameters:
- msg: str, quit message
"""
def disconnect(self, msg: str = "I'll be back!"):
"""
Disconnect from server (will attempt reconnection).
Parameters:
- msg: str, quit message
"""
def jump_server(self, msg: str = "Changing servers"):
"""
Switch to next server in server list.
Parameters:
- msg: str, quit message for current server
"""
@staticmethod
def get_version() -> str:
"""Get IRC library version string."""
def on_ctcp(self, connection, event):
"""
Handle CTCP queries with standard responses.
Automatically responds to VERSION, PING, and TIME queries.
"""
def on_dccchat(self, connection, event):
"""
Handle DCC chat requests.
Override this method to implement custom DCC chat handling.
"""Channel Management
Channel class that tracks channel state including users, modes, and channel properties.
class Channel:
@property
def user_modes(self) -> dict:
"""Dictionary mapping users to their modes in channel."""
@property
def mode_users(self) -> dict:
"""Dictionary mapping modes to users having those modes."""
@property
def modes(self) -> dict:
"""Dictionary of channel modes and their values."""
def users(self) -> list:
"""
Get list of all users in channel.
Returns:
List of user nicknames
"""
def add_user(self, nick: str):
"""
Add user to channel.
Parameters:
- nick: str, user nickname
"""
def remove_user(self, nick: str):
"""
Remove user from channel.
Parameters:
- nick: str, user nickname
"""
def change_nick(self, before: str, after: str):
"""
Handle user nickname change.
Parameters:
- before: str, old nickname
- after: str, new nickname
"""
def has_user(self, nick: str) -> bool:
"""
Check if user is in channel.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is in channel
"""
def opers(self) -> list:
"""Get list of channel operators."""
def voiced(self) -> list:
"""Get list of voiced users."""
def owners(self) -> list:
"""Get list of channel owners."""
def halfops(self) -> list:
"""Get list of half-operators."""
def admins(self) -> list:
"""Get list of channel admins."""
def is_oper(self, nick: str) -> bool:
"""
Check if user is channel operator.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is operator
"""
def is_voiced(self, nick: str) -> bool:
"""
Check if user is voiced.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is voiced
"""
def is_owner(self, nick: str) -> bool:
"""
Check if user is channel owner.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is owner
"""
def is_halfop(self, nick: str) -> bool:
"""
Check if user is half-operator.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is half-op
"""
def is_admin(self, nick: str) -> bool:
"""
Check if user is channel admin.
Parameters:
- nick: str, user nickname
Returns:
bool, True if user is admin
"""
def set_mode(self, mode: str, value=None):
"""
Set channel mode.
Parameters:
- mode: str, mode character
- value: optional mode value/parameter
"""
def clear_mode(self, mode: str, value=None):
"""
Clear channel mode.
Parameters:
- mode: str, mode character
- value: optional mode value/parameter
"""
def has_mode(self, mode: str) -> bool:
"""
Check if channel has mode set.
Parameters:
- mode: str, mode character
Returns:
bool, True if mode is set
"""
def is_moderated(self) -> bool:
"""Check if channel is moderated (+m)."""
def is_secret(self) -> bool:
"""Check if channel is secret (+s)."""
def is_protected(self) -> bool:
"""Check if channel is protected (+t topic lock)."""
def has_topic_lock(self) -> bool:
"""Check if channel has topic lock (+t)."""
def is_invite_only(self) -> bool:
"""Check if channel is invite-only (+i)."""
def has_allow_external_messages(self) -> bool:
"""Check if channel allows external messages (+n disabled)."""
def has_limit(self) -> bool:
"""Check if channel has user limit (+l)."""
def limit(self) -> int:
"""
Get channel user limit.
Returns:
int, user limit or None if not set
"""
def has_key(self) -> bool:
"""Check if channel has key/password (+k)."""Server Configuration
ServerSpec class for defining IRC server connection parameters.
class ServerSpec:
def __init__(self, host: str, port: int = 6667, password: str = None):
"""
Initialize server specification.
Parameters:
- host: str, server hostname
- port: int, server port (default 6667)
- password: str, optional server password
"""
@property
def host(self) -> str:
"""Server hostname."""
@property
def port(self) -> int:
"""Server port."""
@property
def password(self) -> str:
"""Server password (may be None)."""
@classmethod
def ensure(cls, input):
"""
Ensure input is ServerSpec instance.
Parameters:
- input: ServerSpec, dict, or tuple to convert
Returns:
ServerSpec instance
"""Reconnection Strategies
Abstract base class and implementations for automatic reconnection handling.
class ReconnectStrategy:
def run(self, bot):
"""
Execute reconnection strategy.
Parameters:
- bot: SingleServerIRCBot, bot instance to reconnect
"""
class ExponentialBackoff(ReconnectStrategy):
def __init__(self, min_interval: float = 1, max_interval: float = 300):
"""
Initialize exponential backoff reconnection strategy.
Parameters:
- min_interval: float, minimum wait time in seconds
- max_interval: float, maximum wait time in seconds
"""
@property
def min_interval(self) -> float:
"""Minimum reconnection interval."""
@property
def max_interval(self) -> float:
"""Maximum reconnection interval."""
@property
def attempt_count(self) -> int:
"""Number of reconnection attempts made."""
def run(self, bot):
"""
Execute exponential backoff reconnection.
Parameters:
- bot: SingleServerIRCBot, bot to reconnect
"""
def check(self):
"""Check if reconnection should proceed."""Usage Examples
Basic IRC Bot
from irc.bot import SingleServerIRCBot
class MyBot(SingleServerIRCBot):
def __init__(self, channels, nickname, server, port=6667):
# Server specification
server_spec = [{"host": server, "port": port}]
super().__init__(server_spec, nickname, nickname)
self.channels_to_join = channels
def on_welcome(self, connection, event):
"""Called when successfully connected to server."""
for channel in self.channels_to_join:
connection.join(channel)
print(f"Joining {channel}")
def on_pubmsg(self, connection, event):
"""Called when public message received in channel."""
channel = event.target
nick = event.source.nick
message = event.arguments[0]
print(f"[{channel}] <{nick}> {message}")
# Respond to commands
if message.startswith("!hello"):
connection.privmsg(channel, f"Hello {nick}!")
elif message.startswith("!users"):
users = self.channels[channel].users()
connection.privmsg(channel, f"Users in {channel}: {', '.join(users)}")
def on_privmsg(self, connection, event):
"""Called when private message received."""
nick = event.source.nick
message = event.arguments[0]
print(f"PM from {nick}: {message}")
connection.privmsg(nick, f"You said: {message}")
def on_join(self, connection, event):
"""Called when someone joins a channel."""
channel = event.target
nick = event.source.nick
if nick == connection.get_nickname():
print(f"Successfully joined {channel}")
else:
connection.privmsg(channel, f"Welcome {nick}!")
def on_part(self, connection, event):
"""Called when someone leaves a channel."""
channel = event.target
nick = event.source.nick
print(f"{nick} left {channel}")
# Create and start bot
bot = MyBot(["#test", "#botchannel"], "MyBot", "irc.libera.chat")
bot.start()Advanced Bot with Custom Reconnection
from irc.bot import SingleServerIRCBot, ExponentialBackoff
import time
class AdvancedBot(SingleServerIRCBot):
def __init__(self, channels, nickname, servers):
# Custom reconnection strategy - faster initial reconnects
reconnect_strategy = ExponentialBackoff(min_interval=5, max_interval=60)
super().__init__(
servers,
nickname,
f"{nickname} IRC Bot",
recon=reconnect_strategy
)
self.channels_to_join = channels
self.start_time = time.time()
def on_welcome(self, connection, event):
for channel in self.channels_to_join:
connection.join(channel)
def on_pubmsg(self, connection, event):
message = event.arguments[0]
channel = event.target
nick = event.source.nick
if message.startswith("!uptime"):
uptime = int(time.time() - self.start_time)
hours, remainder = divmod(uptime, 3600)
minutes, seconds = divmod(remainder, 60)
connection.privmsg(channel, f"Uptime: {hours}h {minutes}m {seconds}s")
elif message.startswith("!channels"):
channels = list(self.channels.keys())
connection.privmsg(channel, f"I'm in: {', '.join(channels)}")
elif message.startswith("!ops"):
if channel in self.channels:
ops = self.channels[channel].opers()
connection.privmsg(channel, f"Operators: {', '.join(ops)}")
def on_disconnect(self, connection, event):
"""Called when disconnected from server."""
print("Disconnected from server, will attempt reconnection...")
def on_nicknameinuse(self, connection, event):
"""Called when nickname is already in use."""
connection.nick(connection.get_nickname() + "_")
# Multiple server configuration with fallbacks
servers = [
{"host": "irc.libera.chat", "port": 6667},
{"host": "irc.oftc.net", "port": 6667}
]
bot = AdvancedBot(["#test"], "AdvancedBot", servers)
bot.start()Bot with Channel Management
from irc.bot import SingleServerIRCBot
class ChannelBot(SingleServerIRCBot):
def __init__(self, channels, nickname, server, port=6667):
server_spec = [{"host": server, "port": port}]
super().__init__(server_spec, nickname, nickname)
self.channels_to_join = channels
self.admin_users = {"admin_nick"} # Set of admin nicknames
def on_welcome(self, connection, event):
for channel in self.channels_to_join:
connection.join(channel)
def on_pubmsg(self, connection, event):
message = event.arguments[0]
channel = event.target
nick = event.source.nick
# Admin commands
if nick in self.admin_users:
if message.startswith("!kick "):
target = message.split()[1]
if len(message.split()) > 2:
reason = " ".join(message.split()[2:])
else:
reason = "Requested by admin"
connection.kick(channel, target, reason)
elif message.startswith("!mode "):
mode_command = message[6:] # Remove "!mode "
connection.mode(channel, mode_command)
elif message.startswith("!topic "):
new_topic = message[7:] # Remove "!topic "
connection.topic(channel, new_topic)
# Public commands
if message.startswith("!whoami"):
channel_obj = self.channels.get(channel)
if channel_obj:
modes = []
if channel_obj.is_oper(nick):
modes.append("operator")
if channel_obj.is_voiced(nick):
modes.append("voiced")
if channel_obj.is_owner(nick):
modes.append("owner")
if modes:
connection.privmsg(channel, f"{nick}: You are {', '.join(modes)}")
else:
connection.privmsg(channel, f"{nick}: You are a regular user")
elif message.startswith("!count"):
channel_obj = self.channels.get(channel)
if channel_obj:
user_count = len(channel_obj.users())
connection.privmsg(channel, f"Users in {channel}: {user_count}")
def on_mode(self, connection, event):
"""Called when channel or user mode changes."""
target = event.target
modes = event.arguments[0]
print(f"Mode change on {target}: {modes}")
def on_kick(self, connection, event):
"""Called when someone is kicked from channel."""
channel = event.target
kicked_nick = event.arguments[0]
kicker = event.source.nick
reason = event.arguments[1] if len(event.arguments) > 1 else "No reason"
print(f"{kicked_nick} was kicked from {channel} by {kicker}: {reason}")
# If we were kicked, try to rejoin
if kicked_nick == connection.get_nickname():
connection.join(channel)
bot = ChannelBot(["#mychannel"], "ChannelBot", "irc.libera.chat")
bot.start()Multi-Server Bot with State Persistence
import json
import os
from irc.bot import SingleServerIRCBot, ExponentialBackoff
class PersistentBot(SingleServerIRCBot):
def __init__(self, channels, nickname, servers, state_file="bot_state.json"):
super().__init__(servers, nickname, nickname)
self.channels_to_join = channels
self.state_file = state_file
self.user_data = self.load_state()
def load_state(self):
"""Load bot state from file."""
if os.path.exists(self.state_file):
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def save_state(self):
"""Save bot state to file."""
with open(self.state_file, 'w') as f:
json.dump(self.user_data, f, indent=2)
def on_welcome(self, connection, event):
for channel in self.channels_to_join:
connection.join(channel)
def on_pubmsg(self, connection, event):
message = event.arguments[0]
channel = event.target
nick = event.source.nick
if message.startswith("!remember "):
key_value = message[10:].split("=", 1)
if len(key_value) == 2:
key, value = key_value
if nick not in self.user_data:
self.user_data[nick] = {}
self.user_data[nick][key.strip()] = value.strip()
self.save_state()
connection.privmsg(channel, f"{nick}: Remembered {key} = {value}")
elif message.startswith("!recall "):
key = message[8:].strip()
if nick in self.user_data and key in self.user_data[nick]:
value = self.user_data[nick][key]
connection.privmsg(channel, f"{nick}: {key} = {value}")
else:
connection.privmsg(channel, f"{nick}: I don't remember '{key}'")
elif message.startswith("!forget "):
key = message[8:].strip()
if nick in self.user_data and key in self.user_data[nick]:
del self.user_data[nick][key]
self.save_state()
connection.privmsg(channel, f"{nick}: Forgot '{key}'")
elif message.startswith("!list"):
if nick in self.user_data and self.user_data[nick]:
keys = list(self.user_data[nick].keys())
connection.privmsg(channel, f"{nick}: I remember: {', '.join(keys)}")
else:
connection.privmsg(channel, f"{nick}: I don't remember anything for you")
def on_disconnect(self, connection, event):
print("Disconnected, saving state...")
self.save_state()
# Multiple servers with SSL
servers = [
{"host": "irc.libera.chat", "port": 6697, "ssl": True},
{"host": "irc.oftc.net", "port": 6697, "ssl": True}
]
bot = PersistentBot(["#test", "#bots"], "PersistentBot", servers)
try:
bot.start()
except KeyboardInterrupt:
print("Bot shutting down...")
bot.save_state()
bot.die("Bot shutdown requested")Install with Tessl CLI
npx tessl i tessl/pypi-irc