Skip to content

irc_channel

irc_channel

IRCChannel — IRC adapter using stdlib socket.

Classes

IRCChannel 

IRCChannel(server: str = '', *, port: int = 6667, nick: str = '', password: str = '', use_tls: bool = False, bus: Optional[EventBus] = None)

Bases: BaseChannel

IRC channel adapter using stdlib socket (send-only).

PARAMETER DESCRIPTION
server

IRC server hostname. Falls back to IRC_SERVER env var.

TYPE: str DEFAULT: ''

port

IRC server port (default 6667). Falls back to IRC_PORT env var.

TYPE: int DEFAULT: 6667

nick

IRC nickname. Falls back to IRC_NICK env var.

TYPE: str DEFAULT: ''

password

Optional server password. Falls back to IRC_PASSWORD env var.

TYPE: str DEFAULT: ''

use_tls

Whether to use TLS for the connection (default False).

TYPE: bool DEFAULT: False

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/openjarvis/channels/irc_channel.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    server: str = "",
    *,
    port: int = 6667,
    nick: str = "",
    password: str = "",
    use_tls: bool = False,
    bus: Optional[EventBus] = None,
) -> None:
    self._server = server or os.environ.get("IRC_SERVER", "")
    self._port = int(os.environ.get("IRC_PORT", str(port)))
    self._nick = nick or os.environ.get("IRC_NICK", "")
    self._password = password or os.environ.get("IRC_PASSWORD", "")
    self._use_tls = use_tls
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Functions
connect 
connect() -> None

Mark as connected (send-only — no persistent connection).

Source code in src/openjarvis/channels/irc_channel.py 
65
66
67
68
69
70
71
72
73
74
75
def connect(self) -> None:
    """Mark as connected (send-only — no persistent connection)."""
    if not self._server:
        logger.warning("No IRC server configured")
        self._status = ChannelStatus.ERROR
        return
    if not self._nick:
        logger.warning("No IRC nick configured")
        self._status = ChannelStatus.ERROR
        return
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/openjarvis/channels/irc_channel.py 
77
78
79
def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED
send 
send(channel: str, content: str, *, conversation_id: str = '', metadata: Dict[str, Any] | None = None) -> bool

Send a PRIVMSG to an IRC channel via a new socket connection.

Source code in src/openjarvis/channels/irc_channel.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a PRIVMSG to an IRC channel via a new socket connection."""
    if not self._server or not self._nick:
        logger.warning("Cannot send: IRC server or nick not configured")
        return False

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if self._use_tls:
            ctx = ssl.create_default_context()
            sock = ctx.wrap_socket(sock, server_hostname=self._server)

        sock.connect((self._server, self._port))

        if self._password:
            sock.sendall(f"PASS {self._password}\r\n".encode())
        sock.sendall(f"NICK {self._nick}\r\n".encode())
        sock.sendall(f"USER {self._nick} 0 * :{self._nick}\r\n".encode())
        sock.sendall(f"PRIVMSG {channel} :{content}\r\n".encode())
        sock.sendall(b"QUIT\r\n")
        sock.close()

        self._publish_sent(channel, content, conversation_id)
        return True
    except Exception:
        logger.debug("IRC send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/irc_channel.py 
118
119
120
def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status
list_channels 
list_channels() -> List[str]

Return available channel identifiers.

Source code in src/openjarvis/channels/irc_channel.py 
122
123
124
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["irc"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/openjarvis/channels/irc_channel.py 
126
127
128
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)