"""IRC MCP server for Claude Code. Keeps one IRC connection open (TLS), answers PINGs, buffers recent channel messages, and exposes tools to join/leave/send/read/list users. It never sends chat text on its own: the only things it writes to the server unprompted are protocol traffic (NICK/USER registration, PONG replies, keepalive PINGs, and re-joining channels you already joined after a reconnect). """ import asyncio import json import logging import os import random import re import ssl import sys from collections import deque from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path from mcp.server.fastmcp import FastMCP # stdout is the MCP transport, so all logging must go to stderr. logging.basicConfig(stream=sys.stderr, level=logging.INFO, format="%(asctime)s irc-mcp %(levelname)s %(message)s") log = logging.getLogger("irc-mcp") CONFIG_PATH = Path(os.environ.get("IRC_MCP_CONFIG", Path(__file__).with_name("config.json"))) CHANNEL_RE = re.compile(r"^[#&][^\s,\x07]{1,49}$") MAX_LINES_PER_SEND = 10 MAX_PAYLOAD_BYTES = 400 # leaves room for the ":nick!user@host PRIVMSG #chan :" prefix NICK_PREFIXES = "~&@%+" def load_config() -> dict: """Read config.json fresh each time so edits apply without a restart.""" with open(CONFIG_PATH, encoding="utf-8") as f: cfg = json.load(f) cfg.setdefault("port", 6697) cfg.setdefault("tls", True) cfg.setdefault("fallback_nicks", []) cfg.setdefault("username", cfg["nick"]) cfg.setdefault("realname", cfg["nick"]) cfg.setdefault("allowed_channels", []) cfg.setdefault("buffer_size", 200) return cfg def irc_lower(s: str) -> str: """RFC 1459 case folding, used for channel and nick comparison.""" return s.lower().translate(str.maketrans("[]\\~", "{}|^")) def clean(text: str) -> str: """Strip characters that could inject extra IRC commands.""" return text.replace("\r", " ").replace("\n", " ").replace("\x00", "") def parse_line(line: str): """Parse a raw IRC line into (prefix_nick, command, params).""" if line.startswith("@"): # IRCv3 message tags, not needed _, _, line = line.partition(" ") prefix = "" if line.startswith(":"): prefix, _, line = line[1:].partition(" ") trailing = None if " :" in line: line, trailing = line.split(" :", 1) elif line.startswith(":"): line, trailing = "", line[1:] params = line.split() command = params.pop(0).upper() if params else "" if trailing is not None: params.append(trailing) return prefix.split("!", 1)[0], command, params def split_utf8(text: str, limit: int): """Split text into chunks of at most `limit` UTF-8 bytes without breaking characters.""" chunk, size = [], 0 for ch in text: n = len(ch.encode("utf-8")) if size + n > limit: yield "".join(chunk) chunk, size = [], 0 chunk.append(ch) size += n if chunk: yield "".join(chunk) def now() -> str: return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S") class IRCClient: def __init__(self): self.writer: asyncio.StreamWriter | None = None self.nick = "" self.status = "starting" self.registered = asyncio.Event() self.joined: dict[str, set[str]] = {} # folded channel -> nicks self.display: dict[str, str] = {} # folded channel -> name as joined self.wanted: set[str] = set() # channels to re-join after reconnect self.buffers: dict[str, deque] = {} self.buffer_size = 200 self._join_waiters: dict[str, asyncio.Future] = {} self._names_pending: dict[str, set[str]] = {} self._names_waiters: dict[str, list[asyncio.Future]] = {} self._nick_queue: list[str] = [] self._send_lock = asyncio.Lock() # ---------- connection ---------- async def run(self): backoff = 5 while True: try: await self._session() except asyncio.CancelledError: await self._quit() raise except Exception as e: # noqa: BLE001 - keep reconnecting on any failure log.warning("connection lost: %s", e) self.status = f"disconnected ({e}); retrying in {backoff}s" if self.registered.is_set(): backoff = 5 self._reset_state() await asyncio.sleep(backoff) backoff = min(backoff * 2, 300) def _reset_state(self): self.registered.clear() self.writer = None self.joined.clear() for fut in self._join_waiters.values(): if not fut.done(): fut.set_exception(ConnectionError("disconnected")) self._join_waiters.clear() self._names_pending.clear() for futs in self._names_waiters.values(): for fut in futs: if not fut.done(): fut.set_result(None) self._names_waiters.clear() async def _session(self): cfg = load_config() self.buffer_size = int(cfg["buffer_size"]) host, port = cfg["server"], int(cfg["port"]) ctx = ssl.create_default_context() if cfg["tls"] else None self.status = f"connecting to {host}:{port}" log.info(self.status) reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port, ssl=ctx, server_hostname=host if ctx else None), timeout=30) self.writer = writer self._nick_queue = [cfg["nick"], *cfg["fallback_nicks"]] self.nick = self._nick_queue.pop(0) self.status = "registering" await self.send_raw(f"NICK {self.nick}") await self.send_raw(f"USER {cfg['username']} 0 * :{cfg['realname']}") awaiting_pong = False while True: try: raw = await asyncio.wait_for(reader.readline(), timeout=180) except asyncio.TimeoutError: if awaiting_pong: raise ConnectionError("ping timeout") await self.send_raw("PING :keepalive") awaiting_pong = True continue if not raw: raise ConnectionError("server closed the connection") awaiting_pong = False line = raw.decode("utf-8", errors="replace").rstrip("\r\n") if line: await self._handle(line) async def _quit(self): if self.writer: try: await self.send_raw("QUIT :bye") self.writer.close() except Exception: # noqa: BLE001 pass async def send_raw(self, line: str): if not self.writer: raise ConnectionError("not connected to IRC") async with self._send_lock: self.writer.write((clean(line) + "\r\n").encode("utf-8")) await self.writer.drain() # ---------- incoming ---------- def _record(self, channel: str, kind: str, nick: str, text: str = ""): key = irc_lower(channel) buf = self.buffers.get(key) if buf is None or buf.maxlen != self.buffer_size: buf = self.buffers[key] = deque(buf or (), maxlen=self.buffer_size) buf.append({"time": now(), "kind": kind, "nick": nick, "text": text}) def _is_me(self, nick: str) -> bool: return irc_lower(nick) == irc_lower(self.nick) async def _handle(self, line: str): src, cmd, p = parse_line(line) if cmd == "PING": await self.send_raw(f"PONG :{p[0] if p else ''}") elif cmd == "ERROR": raise ConnectionError(p[0] if p else "server ERROR") elif cmd == "001": self.nick = p[0] self.registered.set() self.status = f"connected as {self.nick}" log.info(self.status) for ch in list(self.wanted): await self.send_raw(f"JOIN {self.display.get(ch, ch)}") elif cmd in ("432", "433", "436") and not self.registered.is_set(): if self._nick_queue: self.nick = self._nick_queue.pop(0) else: base = load_config()["nick"][:24] self.nick = f"{base}{random.randint(100, 999)}" log.info("nick taken, trying %s", self.nick) await self.send_raw(f"NICK {self.nick}") elif cmd == "JOIN" and p: ch = irc_lower(p[0]) if self._is_me(src): self.joined[ch] = set() self.display[ch] = p[0] self.wanted.add(ch) fut = self._join_waiters.pop(ch, None) if fut and not fut.done(): fut.set_result(True) else: self.joined.setdefault(ch, set()).add(src) self._record(p[0], "join", src) elif cmd == "PART" and p: ch = irc_lower(p[0]) if self._is_me(src): self.joined.pop(ch, None) self.wanted.discard(ch) elif ch in self.joined: self.joined[ch].discard(src) self._record(p[0], "part", src, p[1] if len(p) > 1 else "") elif cmd == "KICK" and len(p) >= 2: ch = irc_lower(p[0]) reason = p[2] if len(p) > 2 else "" if self._is_me(p[1]): self.joined.pop(ch, None) self.wanted.discard(ch) elif ch in self.joined: self.joined[ch].discard(p[1]) self._record(p[0], "kick", src, f"kicked {p[1]}: {reason}") elif cmd == "QUIT": for ch, nicks in self.joined.items(): if src in nicks: nicks.discard(src) self._record(self.display.get(ch, ch), "quit", src, p[0] if p else "") elif cmd == "NICK" and p: if self._is_me(src): self.nick = p[0] for ch, nicks in self.joined.items(): if src in nicks: nicks.discard(src) nicks.add(p[0]) self._record(self.display.get(ch, ch), "nick", src, f"is now {p[0]}") elif cmd in ("PRIVMSG", "NOTICE") and len(p) >= 2 and p[0][:1] in "#&": text = p[1] kind = "notice" if cmd == "NOTICE" else "message" if text.startswith("\x01ACTION ") and text.endswith("\x01"): kind, text = "action", text[8:-1] elif text.startswith("\x01"): return # other CTCP: ignored, and never answered self._record(p[0], kind, src, text) elif cmd == "353" and len(p) >= 4: # NAMES reply ch = irc_lower(p[2]) self._names_pending.setdefault(ch, set()).update( n.lstrip(NICK_PREFIXES) for n in p[3].split()) elif cmd == "366" and len(p) >= 2: # end of NAMES ch = irc_lower(p[1]) names = self._names_pending.pop(ch, set()) if ch in self.joined: self.joined[ch] = names for fut in self._names_waiters.pop(ch, []): if not fut.done(): fut.set_result(names) elif cmd in ("403", "405", "471", "473", "474", "475", "477") and len(p) >= 2: fut = self._join_waiters.pop(irc_lower(p[1]), None) if fut and not fut.done(): fut.set_exception(RuntimeError(f"{p[1]}: {p[-1]}")) # ---------- actions used by tools ---------- async def ready(self, timeout: float = 20): try: await asyncio.wait_for(self.registered.wait(), timeout) except asyncio.TimeoutError: raise RuntimeError(f"Not connected to IRC yet (status: {self.status})") from None async def join(self, channel: str): await self.ready() ch = irc_lower(channel) if ch in self.joined: return f"Already in {self.display[ch]}." fut = asyncio.get_running_loop().create_future() self._join_waiters[ch] = fut await self.send_raw(f"JOIN {channel}") try: await asyncio.wait_for(fut, 15) except asyncio.TimeoutError: self._join_waiters.pop(ch, None) raise RuntimeError(f"No response from server joining {channel}") from None return f"Joined {channel} as {self.nick}." async def part(self, channel: str): await self.ready() ch = irc_lower(channel) self.wanted.discard(ch) if ch not in self.joined: return f"Not in {channel}." await self.send_raw(f"PART {self.display.get(ch, channel)}") self.joined.pop(ch, None) return f"Left {channel}." async def say(self, channel: str, message: str): await self.ready() ch = irc_lower(channel) if ch not in self.joined: raise RuntimeError(f"Not in {channel}; join it first.") lines = [l for l in message.replace("\r\n", "\n").split("\n") if l.strip()] chunks = [c for l in lines for c in split_utf8(clean(l), MAX_PAYLOAD_BYTES)] if not chunks: raise RuntimeError("Message is empty.") if len(chunks) > MAX_LINES_PER_SEND: raise RuntimeError(f"Message would be {len(chunks)} IRC lines; " f"limit is {MAX_LINES_PER_SEND} to avoid flooding.") target = self.display.get(ch, channel) for i, chunk in enumerate(chunks): if i: await asyncio.sleep(1) # gentle pacing so the server doesn't flag a flood await self.send_raw(f"PRIVMSG {target} :{chunk}") self._record(target, "message", self.nick, chunk) return f"Sent {len(chunks)} line(s) to {target}." async def names(self, channel: str): await self.ready() ch = irc_lower(channel) if ch not in self.joined: raise RuntimeError(f"Not in {channel}; join it first.") fut = asyncio.get_running_loop().create_future() self._names_waiters.setdefault(ch, []).append(fut) await self.send_raw(f"NAMES {self.display.get(ch, channel)}") try: await asyncio.wait_for(fut, 10) except asyncio.TimeoutError: pass # fall back to the list tracked from JOIN/PART events return sorted(self.joined.get(ch, set()), key=str.lower) irc = IRCClient() @asynccontextmanager async def lifespan(_server): task = asyncio.create_task(irc.run()) try: yield finally: task.cancel() try: await task except (asyncio.CancelledError, Exception): # noqa: BLE001 pass mcp = FastMCP("irc", lifespan=lifespan) def check_channel(channel: str, require_allowed: bool = True) -> str: channel = channel.strip() if not CHANNEL_RE.match(channel): raise ValueError(f"'{channel}' is not a valid channel name (it must start with #).") if require_allowed: allowed = {irc_lower(c) for c in load_config()["allowed_channels"]} if irc_lower(channel) not in allowed: raise ValueError(f"{channel} is not in allowed_channels in {CONFIG_PATH}.") return channel @mcp.tool() async def irc_status() -> str: """Show IRC connection status, current nick, joined channels and allowed channels.""" joined = ", ".join(irc.display.get(c, c) for c in irc.joined) or "none" allowed = ", ".join(load_config()["allowed_channels"]) or "none" return f"Status: {irc.status}\nNick: {irc.nick}\nJoined: {joined}\nAllowed: {allowed}" @mcp.tool() async def join_channel(channel: str) -> str: """Join an IRC channel. Only channels listed in allowed_channels in config.json are permitted.""" return await irc.join(check_channel(channel)) @mcp.tool() async def leave_channel(channel: str) -> str: """Leave (part) an IRC channel.""" return await irc.part(check_channel(channel, require_allowed=False)) @mcp.tool() async def send_message(channel: str, message: str) -> str: """Send a message to an IRC channel. Only use this when the user explicitly asks to post something; never send messages on your own initiative.""" return await irc.say(check_channel(channel), message) @mcp.tool() async def read_messages(channel: str, count: int = 20, include_events: bool = False) -> str: """Read the last `count` messages buffered for a channel (only messages seen since joining in this session). Set include_events to also show joins/parts/quits/etc.""" channel = check_channel(channel, require_allowed=False) buf = irc.buffers.get(irc_lower(channel)) if not buf: return f"No messages buffered for {channel}." kinds = None if include_events else {"message", "action", "notice"} items = [m for m in buf if kinds is None or m["kind"] in kinds] items = items[-max(1, min(count, len(items) or 1)):] if not items: return f"No messages buffered for {channel}." out = [] for m in items: if m["kind"] == "message": out.append(f"[{m['time']}] <{m['nick']}> {m['text']}") elif m["kind"] == "action": out.append(f"[{m['time']}] * {m['nick']} {m['text']}") elif m["kind"] == "notice": out.append(f"[{m['time']}] -{m['nick']}- {m['text']}") else: out.append(f"[{m['time']}] -- {m['nick']} {m['kind']} {m['text']}".rstrip()) return "\n".join(out) @mcp.tool() async def list_users(channel: str) -> str: """List the users currently in an IRC channel the bot has joined.""" channel = check_channel(channel, require_allowed=False) users = await irc.names(channel) return f"{len(users)} user(s) in {channel}: " + ", ".join(users) if __name__ == "__main__": mcp.run()