From c50f770f77af2a789e95aa53b66d5deec63be10e Mon Sep 17 00:00:00 2001 From: weishaupt Date: Sun, 11 Oct 2026 10:30:37 -0400 Subject: [PATCH] Add IRC MCP server for Claude Code Connects to Rizon over TLS, answers PINGs, buffers recent channel messages, and exposes join/leave/send/read/list-users tools limited to channels in config.json. Never posts unprompted. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 3 + README.md | 32 ++++ config.json | 11 ++ requirements.txt | 1 + server.py | 460 +++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 507 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.json create mode 100644 requirements.txt create mode 100644 server.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..77ac754 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.venv/ +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..f6cd8d4 --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# irc-mcp + +A small MCP server that gives Claude Code an IRC connection. + +- Connects to `irc.rizon.net:6697` over TLS as `tpuahsiew_claude`, and uses a fallback nick if that one is taken +- Answers server PINGs, sends its own keepalive PING, and reconnects automatically +- Can only join channels listed in `allowed_channels` in `config.json` (re-read on every call, so edits apply right away) +- Keeps a rolling buffer of recent messages per joined channel (`buffer_size`, default 200) +- Never posts on its own. It sends chat only through `send_message`, and does not answer CTCP + +## Tools + +| Tool | What it does | +|---|---| +| `irc_status` | Connection state, nick, joined and allowed channels | +| `join_channel` | Join an allowed channel | +| `leave_channel` | Part a channel | +| `send_message` | Post to a joined channel (max 10 lines, paced 1/s) | +| `read_messages` | Last N buffered messages (`include_events` adds joins, parts, etc.) | +| `list_users` | Nicks currently in a joined channel | + +The buffer only holds messages seen since the bot joined in the current Claude Code session. + +## Setup (Windows) + +```powershell +python -m venv .venv +.venv\Scripts\python.exe -m pip install -r requirements.txt +claude mcp add --scope user irc -- C:\wallpapers\irc-mcp\.venv\Scripts\python.exe C:\wallpapers\irc-mcp\server.py +``` + +Set `IRC_MCP_CONFIG` to use a config file somewhere else. diff --git a/config.json b/config.json new file mode 100644 index 0000000..1322033 --- /dev/null +++ b/config.json @@ -0,0 +1,11 @@ +{ + "server": "irc.rizon.net", + "port": 6697, + "tls": true, + "nick": "tpuahsiew_claude", + "fallback_nicks": ["tpuahsiew_claude_", "tpuahsiew_claude2"], + "username": "claude", + "realname": "Claude Code IRC MCP", + "allowed_channels": ["#blog"], + "buffer_size": 200 +} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f92ca25 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +mcp>=1.10,<2 diff --git a/server.py b/server.py new file mode 100644 index 0000000..1e2b613 --- /dev/null +++ b/server.py @@ -0,0 +1,460 @@ +"""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()