tessl/pypi-irc
IRC (Internet Relay Chat) protocol library for Python
—
Pending
synchronous-client.mddocs/
Synchronous IRC Client
Core IRC client functionality using traditional socket connections with select-based event processing. The synchronous client provides the foundation for IRC protocol communication with blocking operations and callback-based event handling.
Capabilities
Reactor
Central event loop that manages IRC connections and dispatches events to handlers. Uses select() for socket monitoring and provides the main event processing infrastructure.
class Reactor:
def __init__(self, on_connect=None, on_disconnect=None):
"""
Initialize IRC reactor for event processing.
Parameters:
- on_connect: callable, called when connection established
- on_disconnect: callable, called when connection lost
"""
def server(self) -> ServerConnection:
"""
Create a new ServerConnection for IRC server communication.
Returns:
ServerConnection instance
"""
def dcc(self, dcctype: str = "chat") -> DCCConnection:
"""
Create a DCC connection for direct client communication.
Parameters:
- dcctype: str, type of DCC connection ("chat" or "raw")
Returns:
DCCConnection instance
"""
def process_once(self, timeout: float = 0):
"""
Process events once with optional timeout.
Parameters:
- timeout: float, timeout in seconds (0 for non-blocking)
"""
def process_forever(self, timeout: float = 0.2):
"""
Process events indefinitely until disconnection.
Parameters:
- timeout: float, timeout between iterations in seconds
"""
def add_global_handler(self, event: str, handler, priority: int = 0):
"""
Add global event handler for all connections.
Parameters:
- event: str, event type to handle
- handler: callable, handler function (connection, event)
- priority: int, handler priority (lower = higher priority)
"""
def remove_global_handler(self, event: str, handler):
"""
Remove global event handler.
Parameters:
- event: str, event type
- handler: callable, handler function to remove
"""
def disconnect_all(self, message: str = ""):
"""
Disconnect all managed connections.
Parameters:
- message: str, quit message
"""ServerConnection
Individual IRC server connection handling protocol communication, command sending, and event processing for a single IRC server.
class ServerConnection:
@property
def connected(self) -> bool:
"""Whether connection is established."""
@property
def features(self) -> FeatureSet:
"""Server-announced features and capabilities."""
def connect(self, server: str, port: int, nickname: str, password: str | None = None,
username: str | None = None, ircname: str | None = None,
connect_factory=connection.Factory(), sasl_login: str | None = None):
"""
Connect to IRC server.
Parameters:
- server: str, server hostname
- port: int, server port
- nickname: str, desired nickname
- password: str, optional server password
- username: str, optional username (defaults to nickname)
- ircname: str, optional real name (defaults to nickname)
- connect_factory: Factory, optional connection factory
- sasl_login: str | None, optional SASL username for authentication
"""
def reconnect(self):
"""Reconnect to the same server with same parameters."""
def disconnect(self, message: str = ""):
"""
Disconnect from server.
Parameters:
- message: str, quit message
"""
def close(self):
"""Close connection immediately without sending QUIT."""
def get_server_name(self) -> str:
"""Get connected server name."""
def get_nickname(self) -> str:
"""Get current nickname."""
def is_connected(self) -> bool:
"""Check if connected to server."""
def as_nick(self, name: str):
"""
Context manager for temporary nickname changes.
Parameters:
- name: str, temporary nickname
Usage:
with connection.as_nick("tempnick"):
# nickname is temporarily changed
pass
# nickname is restored
"""
def send_raw(self, string: str):
"""
Send raw IRC command.
Parameters:
- string: str, raw IRC protocol message
"""
def send_items(self, *items):
"""
Send IRC command from items.
Parameters:
- items: IRC command components
"""
def set_rate_limit(self, frequency: float):
"""
Set rate limiting for outgoing messages.
Parameters:
- frequency: float, maximum messages per second
"""
def set_keepalive(self, interval: int):
"""
Set keepalive ping interval.
Parameters:
- interval: int, seconds between keepalive pings
"""
def add_global_handler(self, event: str, handler, priority: int = 0):
"""
Add event handler for this connection.
Parameters:
- event: str, event type
- handler: callable, handler function
- priority: int, handler priority
"""
def remove_global_handler(self, event: str, handler):
"""
Remove event handler from this connection.
Parameters:
- event: str, event type
- handler: callable, handler to remove
"""IRC Protocol Commands
Standard IRC protocol commands for server communication, channel operations, and user interactions.
class ServerConnection:
def nick(self, newnick: str):
"""
Change nickname.
Parameters:
- newnick: str, new nickname
"""
def user(self, username: str, ircname: str):
"""
Send USER command during registration.
Parameters:
- username: str, username
- ircname: str, real name
"""
def join(self, channel: str, key: str = ""):
"""
Join IRC channel.
Parameters:
- channel: str, channel name (with # prefix)
- key: str, optional channel key/password
"""
def part(self, channel: str, message: str = ""):
"""
Leave IRC channel.
Parameters:
- channel: str, channel name
- message: str, optional part message
"""
def privmsg(self, target: str, text: str):
"""
Send private message to user or channel.
Parameters:
- target: str, recipient (nickname or channel)
- text: str, message text
"""
def notice(self, target: str, text: str):
"""
Send notice to user or channel.
Parameters:
- target: str, recipient (nickname or channel)
- text: str, notice text
"""
def quit(self, message: str = ""):
"""
Quit IRC server.
Parameters:
- message: str, quit message
"""
def ping(self, target: str):
"""
Send PING to server or user.
Parameters:
- target: str, ping target
"""
def pong(self, target: str):
"""
Send PONG response.
Parameters:
- target: str, pong target
"""
def kick(self, channel: str, nick: str, comment: str = ""):
"""
Kick user from channel.
Parameters:
- channel: str, channel name
- nick: str, user to kick
- comment: str, optional kick reason
"""
def mode(self, target: str, command: str):
"""
Set user or channel mode.
Parameters:
- target: str, mode target (user or channel)
- command: str, mode command (+/-modes)
"""
def topic(self, channel: str, new_topic: str = None):
"""
Get or set channel topic.
Parameters:
- channel: str, channel name
- new_topic: str, optional new topic (None to query)
"""
def whois(self, target: str):
"""
Query user information.
Parameters:
- target: str, user nickname
"""
def who(self, target: str, op: str = ""):
"""
Query user list.
Parameters:
- target: str, query target (channel or mask)
- op: str, optional WHO flags
"""
def action(self, target: str, action: str):
"""
Send CTCP ACTION command (/me action).
Parameters:
- target: str, channel or user to send action to
- action: str, action message
"""
def admin(self, server: str = ""):
"""
Query server administrative information.
Parameters:
- server: str, optional server name
"""
def cap(self, subcommand: str, *args):
"""
Send CAP command for IRCv3 capability negotiation.
Parameters:
- subcommand: str, CAP subcommand (LS, REQ, ACK, etc.)
- *args: additional arguments for the subcommand
"""
def ctcp(self, ctcptype: str, target: str, parameter: str = ""):
"""
Send CTCP command to target.
Parameters:
- ctcptype: str, CTCP command type (VERSION, TIME, etc.)
- target: str, target user
- parameter: str, optional parameter
"""
def ctcp_reply(self, target: str, parameter: str):
"""
Send CTCP reply.
Parameters:
- target: str, reply target
- parameter: str, reply parameter
"""
def info(self, server: str = ""):
"""
Query server information.
Parameters:
- server: str, optional server name
"""
def invite(self, nick: str, channel: str):
"""
Invite user to channel.
Parameters:
- nick: str, user to invite
- channel: str, target channel
"""
def ison(self, nicks: list):
"""
Check if nicknames are online.
Parameters:
- nicks: list, nicknames to check
"""
def list(self, channels: list = None, server: str = ""):
"""
List channels.
Parameters:
- channels: list, optional channel list to query
- server: str, optional server name
"""
def names(self, channels: list = None):
"""
Get channel member lists.
Parameters:
- channels: list, optional channel list
"""
def oper(self, nick: str, password: str):
"""
Authenticate as IRC operator.
Parameters:
- nick: str, operator nickname
- password: str, operator password
"""
def pass_(self, password: str):
"""
Send server password (PASS command).
Parameters:
- password: str, server password
"""
def privmsg_many(self, targets: list, text: str):
"""
Send message to multiple targets.
Parameters:
- targets: list, target channels/users
- text: str, message text
"""
def stats(self, statstype: str, server: str = ""):
"""
Query server statistics.
Parameters:
- statstype: str, statistics type
- server: str, optional server name
"""
def time(self, server: str = ""):
"""
Query server time.
Parameters:
- server: str, optional server name
"""
def userhost(self, nicks: list):
"""
Get user hostmask information.
Parameters:
- nicks: list, nicknames to query
"""
def version(self, server: str = ""):
"""
Query server version.
Parameters:
- server: str, optional server name
"""
def wallops(self, text: str):
"""
Send operator wall message.
Parameters:
- text: str, message text
"""
def whowas(self, nick: str, max: str = "", server: str = ""):
"""
Query historical user information.
Parameters:
- nick: str, nickname to query
- max: str, optional maximum entries
- server: str, optional server name
"""SimpleIRCClient
High-level IRC client that manages a single server connection with simplified event handling and automatic connection management.
class SimpleIRCClient:
@property
def reactor(self) -> Reactor:
"""Access to underlying reactor."""
@property
def connection(self) -> ServerConnection:
"""Active server connection."""
@property
def dcc_connections(self) -> list:
"""List of active DCC connections."""
def __init__(self):
"""Initialize simple IRC client."""
def connect(self, server: str, port: int, nickname: str, **kwargs):
"""
Connect to IRC server.
Parameters:
- server: str, server hostname
- port: int, server port
- nickname: str, desired nickname
- **kwargs: additional connection parameters
"""
def start(self):
"""Start client event processing loop."""
def dcc(self, dcctype: str = "chat") -> DCCConnection:
"""
Create DCC connection.
Parameters:
- dcctype: str, DCC connection type
Returns:
DCCConnection instance
"""
def dcc_connect(self, address: tuple, port: int, dcctype: str = "chat") -> DCCConnection:
"""
Connect to DCC peer.
Parameters:
- address: tuple, peer address
- port: int, peer port
- dcctype: str, DCC type
Returns:
DCCConnection instance
"""
def dcc_listen(self, dcctype: str = "chat") -> DCCConnection:
"""
Listen for DCC connections.
Parameters:
- dcctype: str, DCC type
Returns:
DCCConnection instance
"""Usage Examples
Basic Connection and Event Handling
import irc.client
def on_connect(connection, event):
print("Connected to server")
connection.join("#mychannel")
def on_join(connection, event):
print(f"Joined {event.target}")
connection.privmsg(event.target, "Hello everyone!")
def on_pubmsg(connection, event):
print(f"<{event.source.nick}> {event.arguments[0]}")
# Create client and connect
client = irc.client.SimpleIRCClient()
client.connect("irc.libera.chat", 6667, "mybot")
# Add event handlers
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("join", on_join)
client.connection.add_global_handler("pubmsg", on_pubmsg)
# Start processing events
client.start()Multiple Connections with Reactor
import irc.client
reactor = irc.client.Reactor()
# Create multiple connections
conn1 = reactor.server()
conn2 = reactor.server()
def on_connect(connection, event):
if connection == conn1:
connection.join("#channel1")
else:
connection.join("#channel2")
reactor.add_global_handler("welcome", on_connect)
# Connect to different servers
conn1.connect("irc.libera.chat", 6667, "bot1")
conn2.connect("irc.oftc.net", 6667, "bot2")
# Process events for all connections
reactor.process_forever()Rate Limiting and Connection Management
import irc.client
client = irc.client.SimpleIRCClient()
client.connect("irc.libera.chat", 6667, "mybot")
# Set rate limiting to 1 message per second
client.connection.set_rate_limit(1.0)
# Set keepalive to ping every 60 seconds
client.connection.set_keepalive(60)
def on_connect(connection, event):
connection.join("#test")
def on_disconnect(connection, event):
print("Disconnected, attempting reconnection...")
connection.reconnect()
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("disconnect", on_disconnect)
client.start()Install with Tessl CLI
npx tessl i tessl/pypi-irc