tessl/pypi-irc
IRC (Internet Relay Chat) protocol library for Python
—
Pending
asynchronous-client.mddocs/
Asynchronous IRC Client
Asyncio-based IRC client implementation providing full async/await support for modern Python applications. Uses asyncio protocols for non-blocking network communication and integrates seamlessly with asyncio event loops.
Capabilities
AioReactor
Asyncio-based reactor for managing IRC connections in asynchronous applications. Integrates with asyncio event loops for non-blocking event processing.
class AioReactor:
def __init__(self, on_connect=None, on_disconnect=None, loop=None):
"""
Initialize asyncio IRC reactor.
Parameters:
- on_connect: callable, called when connection established
- on_disconnect: callable, called when connection lost
- loop: asyncio event loop, uses current loop if None
"""
@property
def loop(self):
"""Asyncio event loop used by reactor."""
def server(self) -> AioConnection:
"""
Create new AioConnection for IRC server communication.
Returns:
AioConnection instance
"""
async def process_forever(self):
"""
Process events indefinitely using asyncio.
This method runs until all connections are closed.
"""
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
"""AioConnection
Asyncio-based IRC server connection using asyncio protocols for non-blocking network communication.
class AioConnection:
@property
def connected(self) -> bool:
"""Whether connection is established."""
@property
def features(self) -> FeatureSet:
"""Server-announced features and capabilities."""
async def connect(self, server, port, nickname, password=None,
username=None, ircname=None, connect_factory=connection.AioFactory()):
"""
Connect to IRC server asynchronously.
Parameters:
- server: server hostname
- port: server port
- nickname: desired nickname
- password: optional server password
- username: optional username (defaults to nickname)
- ircname: optional real name (defaults to nickname)
- connect_factory: optional connection factory (defaults to AioFactory())
"""
def process_data(self, new_data: str):
"""
Process incoming IRC data.
Parameters:
- new_data: str, raw IRC protocol data
"""
def send_raw(self, string: str):
"""
Send raw IRC command.
Parameters:
- string: str, raw IRC protocol message
"""
def disconnect(self, message: str = ""):
"""
Disconnect from server.
Parameters:
- message: str, quit message
"""
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 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
"""AioSimpleIRCClient
High-level asyncio IRC client that manages a single server connection with simplified event handling for async applications.
class AioSimpleIRCClient:
@property
def reactor(self) -> AioReactor:
"""Access to underlying asyncio reactor."""
@property
def connection(self) -> AioConnection:
"""Active server connection."""
def __init__(self):
"""Initialize asyncio simple IRC client."""
async def connect(self, server: str, port: int, nickname: str, **kwargs):
"""
Connect to IRC server asynchronously.
Parameters:
- server: str, server hostname
- port: int, server port
- nickname: str, desired nickname
- **kwargs: additional connection parameters
"""IrcProtocol
Asyncio protocol implementation for IRC communication handling low-level network operations.
class IrcProtocol:
def __init__(self, connection, loop):
"""
Initialize IRC protocol handler.
Parameters:
- connection: AioConnection, associated connection
- loop: asyncio event loop
"""
def data_received(self, data: bytes):
"""
Handle received network data.
Parameters:
- data: bytes, raw network data
"""
def connection_lost(self, exc):
"""
Handle connection loss.
Parameters:
- exc: Exception, optional exception that caused disconnection
"""IRC Protocol Commands (Async)
All standard IRC protocol commands are available on AioConnection with the same interface as ServerConnection:
class AioConnection:
def nick(self, newnick: str):
"""Change nickname."""
def user(self, username: str, ircname: str):
"""Send USER command during registration."""
def join(self, channel: str, key: str = ""):
"""Join IRC channel."""
def part(self, channel: str, message: str = ""):
"""Leave IRC channel."""
def privmsg(self, target: str, text: str):
"""Send private message to user or channel."""
def notice(self, target: str, text: str):
"""Send notice to user or channel."""
def quit(self, message: str = ""):
"""Quit IRC server."""
def ping(self, target: str):
"""Send PING to server or user."""
def pong(self, target: str):
"""Send PONG response."""
def kick(self, channel: str, nick: str, comment: str = ""):
"""Kick user from channel."""
def mode(self, target: str, command: str):
"""Set user or channel mode."""
def topic(self, channel: str, new_topic: str = None):
"""Get or set channel topic."""
def whois(self, target: str):
"""Query user information."""
def who(self, target: str, op: str = ""):
"""Query user list."""Usage Examples
Basic Asyncio IRC Client
import asyncio
from irc.client_aio import AioSimpleIRCClient
async def main():
client = AioSimpleIRCClient()
def on_connect(connection, event):
print("Connected to server")
connection.join("#test")
def on_join(connection, event):
print(f"Joined {event.target}")
connection.privmsg(event.target, "Hello from async bot!")
def on_pubmsg(connection, event):
print(f"<{event.source.nick}> {event.arguments[0]}")
# Echo messages back
connection.privmsg(event.target, f"Echo: {event.arguments[0]}")
# 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)
# Connect to server
await client.connect("irc.libera.chat", 6667, "asyncbot")
# Keep running
await asyncio.Event().wait()
asyncio.run(main())Multiple Async Connections
import asyncio
from irc.client_aio import AioReactor
async def main():
reactor = AioReactor()
# Create multiple connections
conn1 = reactor.server()
conn2 = reactor.server()
def on_connect(connection, event):
if connection == conn1:
connection.join("#channel1")
else:
connection.join("#channel2")
def on_pubmsg(connection, event):
print(f"[{connection.get_server_name()}] <{event.source.nick}> {event.arguments[0]}")
reactor.add_global_handler("welcome", on_connect)
reactor.add_global_handler("pubmsg", on_pubmsg)
# Connect to multiple servers
await conn1.connect("irc.libera.chat", 6667, "bot1")
await conn2.connect("irc.oftc.net", 6667, "bot2")
# Process events for all connections
await reactor.process_forever()
asyncio.run(main())Async Bot with Periodic Tasks
import asyncio
from irc.client_aio import AioSimpleIRCClient
async def periodic_announce(client):
"""Send periodic announcements."""
while True:
await asyncio.sleep(300) # 5 minutes
if client.connection.is_connected():
client.connection.privmsg("#test", "Periodic announcement!")
async def main():
client = AioSimpleIRCClient()
def on_connect(connection, event):
connection.join("#test")
print("Connected and joined channel")
def on_pubmsg(connection, event):
message = event.arguments[0]
if message.startswith("!time"):
import datetime
now = datetime.datetime.now().strftime("%H:%M:%S")
connection.privmsg(event.target, f"Current time: {now}")
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("pubmsg", on_pubmsg)
# Connect to server
await client.connect("irc.libera.chat", 6667, "timebot")
# Start periodic task
announce_task = asyncio.create_task(periodic_announce(client))
try:
# Wait indefinitely
await asyncio.Event().wait()
except KeyboardInterrupt:
announce_task.cancel()
client.connection.quit("Bot shutting down")
asyncio.run(main())Async Connection with SSL
import asyncio
import ssl
from irc.client_aio import AioSimpleIRCClient
from irc.connection import AioFactory
async def main():
# Create SSL context
ssl_context = ssl.create_default_context()
# Create SSL connection factory
ssl_factory = AioFactory(ssl=ssl_context)
client = AioSimpleIRCClient()
def on_connect(connection, event):
connection.join("#secure")
print("Connected via SSL")
client.connection.add_global_handler("welcome", on_connect)
# Connect with SSL on port 6697
await client.connect(
"irc.libera.chat",
6697,
"sslbot",
connect_factory=ssl_factory
)
await asyncio.Event().wait()
asyncio.run(main())Integration with Web Framework
import asyncio
from aiohttp import web
from irc.client_aio import AioSimpleIRCClient
class IRCBot:
def __init__(self):
self.client = AioSimpleIRCClient()
self.messages = []
async def start(self):
def on_connect(connection, event):
connection.join("#webapp")
def on_pubmsg(connection, event):
self.messages.append({
'nick': event.source.nick,
'message': event.arguments[0],
'timestamp': asyncio.get_event_loop().time()
})
# Keep only last 100 messages
self.messages = self.messages[-100:]
self.client.connection.add_global_handler("welcome", on_connect)
self.client.connection.add_global_handler("pubmsg", on_pubmsg)
await self.client.connect("irc.libera.chat", 6667, "webapp_bot")
def send_message(self, message):
self.client.connection.privmsg("#webapp", message)
bot = IRCBot()
async def send_message(request):
data = await request.json()
bot.send_message(data['message'])
return web.json_response({'status': 'sent'})
async def get_messages(request):
return web.json_response({'messages': bot.messages})
async def init_app():
await bot.start()
app = web.Application()
app.router.add_post('/send', send_message)
app.router.add_get('/messages', get_messages)
return app
if __name__ == '__main__':
web.run_app(init_app(), host='0.0.0.0', port=8080)Install with Tessl CLI
npx tessl i tessl/pypi-irc