weishauptandClaude Opus 5.5
fc973790fb
- Nick is now clod with clod_ as fallback; while on a fallback nick the server retries NICK clod every nick_retry_seconds (default 180). - Identify to NickServ whenever holding the configured nick, using the password in nickserv_password.txt (gitignored). - Add #/g/tv to allowed_channels. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
505 lines
19 KiB
Python
505 lines
19 KiB
Python
"""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, NickServ IDENTIFY, periodic NICK attempts to reclaim the
|
|
configured nick, 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)
|
|
cfg.setdefault("nickserv_password_file", "")
|
|
cfg.setdefault("nick_retry_seconds", 180)
|
|
return cfg
|
|
|
|
|
|
def nickserv_password(cfg: dict) -> str:
|
|
"""Read the NickServ password from its (gitignored) file, relative to the config."""
|
|
name = cfg["nickserv_password_file"]
|
|
if not name:
|
|
return ""
|
|
path = CONFIG_PATH.parent / name
|
|
try:
|
|
return path.read_text(encoding="utf-8").strip()
|
|
except FileNotFoundError:
|
|
log.warning("NickServ password file %s not found; not identifying", path)
|
|
return ""
|
|
|
|
|
|
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']}")
|
|
|
|
reclaim = asyncio.create_task(self._reclaim_nick())
|
|
try:
|
|
await self._read_loop(reader)
|
|
finally:
|
|
reclaim.cancel()
|
|
|
|
async def _read_loop(self, reader: asyncio.StreamReader):
|
|
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 _reclaim_nick(self):
|
|
"""While on a fallback nick, periodically try to switch back to the configured one."""
|
|
while True:
|
|
cfg = load_config()
|
|
await asyncio.sleep(max(30, int(cfg["nick_retry_seconds"])))
|
|
cfg = load_config()
|
|
if self.registered.is_set() and not self._is_me(cfg["nick"]):
|
|
log.info("trying to reclaim nick %s", cfg["nick"])
|
|
await self.send_raw(f"NICK {cfg['nick']}")
|
|
|
|
async def _identify(self):
|
|
"""Identify to NickServ, but only while holding the configured (registered) nick."""
|
|
cfg = load_config()
|
|
password = nickserv_password(cfg)
|
|
if password and self._is_me(cfg["nick"]):
|
|
log.info("identifying to NickServ as %s", self.nick)
|
|
await self.send_raw(f"PRIVMSG NickServ :IDENTIFY {password}")
|
|
|
|
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)
|
|
await self._identify()
|
|
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]
|
|
self.status = f"connected as {self.nick}"
|
|
await self._identify()
|
|
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()
|