NNTP MCP-сервер: мониторинг новостных групп и ответы в тредах.
Сценарий: скрипт запускается периодически, scan_group находит новые сообщения с момента прошлого запуска, возвращает треды целиком (с контекстом), LLM принимает решение, post_reply отправляет ответ. Состояние хранится в nntp_mcp_state.json рядом со скриптом.
This commit is contained in:
ysv committed
1 parent
8b4659cffb
commit
367f024cbd
4 files changed
+486
No files matched your search
Whitespace-only changes.
@@ -427,6 +427,72 @@ def get_latest_messages(group_name: str, count: int = 10) -> list[dict]:
|
||||
except Exception as e:
|
||||
return [{"error": f"Ошибка NNTP при получении последних сообщений: {e}"}]
|
||||
|
||||
@mcp.tool()
|
||||
def get_thread_root(message_id: str) -> dict:
|
||||
"""
|
||||
Находит Message-ID первого (корневого) сообщения в ветке обсуждения.
|
||||
|
||||
:param message_id: Message-ID любого сообщения из ветки (например, '<abc@example.com>')
|
||||
:return: Словарь с полями root_message_id, subject, author или error
|
||||
"""
|
||||
try:
|
||||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
|
||||
# Получаем исходное сообщение
|
||||
resp, info = s.article(message_id)
|
||||
raw_bytes = b"\n".join(info.lines)
|
||||
msg = email.message_from_bytes(raw_bytes, policy=default)
|
||||
|
||||
# Проверяем заголовок References - там все Message-ID в цепочке
|
||||
references = msg.get("References", "").strip()
|
||||
|
||||
if references:
|
||||
# References содержит список ID через пробел, первый - корень ветки
|
||||
ref_ids = re.findall(r'<[^>]+>', references)
|
||||
if ref_ids:
|
||||
root_id = ref_ids[0]
|
||||
|
||||
# Получаем информацию о корневом сообщении
|
||||
try:
|
||||
resp_root, info_root = s.article(root_id)
|
||||
raw_root = b"\n".join(info_root.lines)
|
||||
root_msg = email.message_from_bytes(raw_root, policy=default)
|
||||
|
||||
return {
|
||||
"root_message_id": root_id,
|
||||
"subject": decode_mime_header(root_msg.get("Subject", "[Без темы]")),
|
||||
"author": decode_mime_header(root_msg.get("From", "[Неизвестно]")),
|
||||
"date": root_msg.get("Date", "")
|
||||
}
|
||||
except Exception:
|
||||
# Если корневое сообщение недоступно, возвращаем хотя бы его ID
|
||||
return {
|
||||
"root_message_id": root_id,
|
||||
"subject": "[Сообщение недоступно]",
|
||||
"author": "[Неизвестно]",
|
||||
"date": ""
|
||||
}
|
||||
|
||||
# Если References нет, проверяем In-Reply-To и поднимаемся рекурсивно
|
||||
in_reply_to = msg.get("In-Reply-To", "").strip()
|
||||
|
||||
if in_reply_to:
|
||||
# Извлекаем Message-ID из In-Reply-To
|
||||
match = re.search(r'<[^>]+>', in_reply_to)
|
||||
if match:
|
||||
parent_id = match.group(0)
|
||||
# Рекурсивно ищем корень через родительское сообщение
|
||||
return get_thread_root(parent_id)
|
||||
|
||||
# Если ни References, ни In-Reply-To нет - это и есть корень
|
||||
return {
|
||||
"root_message_id": message_id,
|
||||
"subject": decode_mime_header(msg.get("Subject", "[Без темы]")),
|
||||
"author": decode_mime_header(msg.get("From", "[Неизвестно]")),
|
||||
"date": msg.get("Date", "")
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
return {"error": f"Не удалось найти корень ветки для {message_id}: {e}"}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"news.skysamara.ru/sky.test": {
|
||||
"last_seen": null,
|
||||
"my_ids": [
|
||||
"<179302383962.14856.2029190571264960217@test.com>"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,412 @@
|
||||
"""
|
||||
NNTP MCP-сервер: мониторинг новостных групп и ответы в тредах.
|
||||
"""
|
||||
|
||||
import warnings
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# stderr отправляем в лог-файл, чтобы предупреждения и трейсбэки
|
||||
# не ломали stdio-транспорт, но ошибки оставались диагностируемыми
|
||||
sys.stderr = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "nntp_mcp.log"), "a")
|
||||
|
||||
import json
|
||||
import re
|
||||
import datetime
|
||||
import nntplib
|
||||
import email
|
||||
from email.message import EmailMessage
|
||||
from email.header import decode_header
|
||||
from email.utils import parseaddr, formataddr, make_msgid, formatdate, parsedate_to_datetime
|
||||
from email.policy import default as email_policy
|
||||
|
||||
# Автоматическое переключение импорта в зависимости от окружения
|
||||
try:
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
mcp = FastMCP("nntp-monitor")
|
||||
# Используем прокси-декоратор для единообразия регистрации ниже
|
||||
def register_mcp_tool(func):
|
||||
return mcp.tool()(func)
|
||||
except ImportError:
|
||||
from mcp.server.mcpserver import MCPServer
|
||||
mcp = MCPServer("nntp-monitor")
|
||||
def register_mcp_tool(func):
|
||||
if hasattr(mcp, 'list_tools') and hasattr(mcp.list_tools, 'add_tool'):
|
||||
mcp.list_tools.add_tool(func)
|
||||
elif hasattr(mcp, 'register_tool'):
|
||||
mcp.register_tool(func)
|
||||
else:
|
||||
# ИСПРАВЛЕНО: Правильный вызов декоратора со скобками для базового MCPServer
|
||||
decorator = mcp.tool()
|
||||
decorator(func)
|
||||
return func
|
||||
|
||||
STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "nntp_mcp_state.json")
|
||||
DEFAULT_WINDOW = 1000
|
||||
MAX_THREAD_MSGS = 40
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- состояние
|
||||
|
||||
def _state_key(server: str, group: str) -> str:
|
||||
return f"{server.lower()}/{group.lower()}"
|
||||
|
||||
|
||||
def _load_state() -> dict:
|
||||
try:
|
||||
with open(STATE_FILE, encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _save_state(state: dict) -> None:
|
||||
tmp = STATE_FILE + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(state, f, ensure_ascii=False, indent=1)
|
||||
os.replace(tmp, STATE_FILE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- утилиты
|
||||
|
||||
def _decode(header_value) -> str:
|
||||
if not header_value:
|
||||
return ""
|
||||
try:
|
||||
parts = []
|
||||
for text, enc in decode_header(str(header_value)):
|
||||
if isinstance(text, bytes):
|
||||
parts.append(text.decode(enc or "utf-8", errors="ignore"))
|
||||
else:
|
||||
parts.append(str(text))
|
||||
return "".join(parts)
|
||||
except Exception:
|
||||
return str(header_value)
|
||||
|
||||
|
||||
def _parse_date(date_str: str) -> str:
|
||||
if not date_str:
|
||||
return ""
|
||||
try:
|
||||
dt = parsedate_to_datetime(date_str)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=datetime.timezone.utc)
|
||||
return dt.isoformat()
|
||||
except Exception:
|
||||
return str(date_str)
|
||||
|
||||
|
||||
def _clean_quotes(body_text: str) -> str:
|
||||
if not body_text:
|
||||
return ""
|
||||
intro = re.compile(
|
||||
r"on\s+.*?\s+wrote\s*:|в\s+письме\s+от.*?:|писал\s*\(а\)\s*:|"
|
||||
r"---.*?original\s+message.*?---",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
clean, prev_blank = [], False
|
||||
for line in body_text.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith(">"):
|
||||
continue
|
||||
if intro.search(stripped):
|
||||
continue
|
||||
if not stripped and prev_blank:
|
||||
continue
|
||||
prev_blank = not stripped
|
||||
clean.append(line)
|
||||
return "\n".join(clean).strip()
|
||||
|
||||
|
||||
def _extract_body(msg) -> str:
|
||||
body = ""
|
||||
for part in (msg.walk() if msg.is_multipart() else [msg]):
|
||||
if str(part.get("Content-Disposition", "")).lower().startswith("attachment"):
|
||||
continue
|
||||
if part.get_content_type() != "text/plain":
|
||||
continue
|
||||
payload = part.get_payload(decode=True)
|
||||
if payload:
|
||||
body += payload.decode(part.get_content_charset() or "utf-8", errors="ignore")
|
||||
if not body.strip():
|
||||
for part in (msg.walk() if msg.is_multipart() else [msg]):
|
||||
if part.get_content_type() == "text/html":
|
||||
payload = part.get_payload(decode=True)
|
||||
if payload:
|
||||
body = "[HTML]\n" + payload.decode(
|
||||
part.get_content_charset() or "utf-8", errors="ignore")
|
||||
break
|
||||
return body or "[Тело сообщения пустое]"
|
||||
|
||||
|
||||
def _connect(server: str, port: int, user: str, password: str):
|
||||
if port == 563:
|
||||
return nntplib.NNTP_SSL(server, port=port, timeout=30,
|
||||
user=user or None, password=password or None)
|
||||
return nntplib.NNTP(server, port=port, timeout=30,
|
||||
user=user or None, password=password or None)
|
||||
|
||||
|
||||
def _fetch_index(s, last: int, window: int) -> list[dict]:
|
||||
start = max(1, last - window + 1)
|
||||
try:
|
||||
_, overviews = s.over((str(start), str(last)))
|
||||
except Exception:
|
||||
return []
|
||||
index = []
|
||||
for num, over in overviews:
|
||||
msgid = (over.get("message-id") or "").strip() or f"<local-{num}>"
|
||||
index.append({
|
||||
"num": int(num),
|
||||
"msgid": msgid,
|
||||
"subject": _decode(over.get("subject", "")),
|
||||
"from": _decode(over.get("from", "")),
|
||||
"date": _parse_date(over.get("date", "")),
|
||||
"refs": [r for r in (over.get("references") or "").split()
|
||||
if r.startswith("<")],
|
||||
})
|
||||
index.sort(key=lambda m: m["num"])
|
||||
return index
|
||||
|
||||
|
||||
def _make_root_resolver(index_by_id: dict):
|
||||
resolved = {}
|
||||
|
||||
def root_of(msg: dict) -> str:
|
||||
chain, cur = [], msg
|
||||
while True:
|
||||
key = cur["msgid"]
|
||||
if key in resolved:
|
||||
r = resolved[key]
|
||||
break
|
||||
if key in chain:
|
||||
r = key
|
||||
break
|
||||
chain.append(key)
|
||||
parent_id = cur["refs"][0] if cur["refs"] else None
|
||||
parent = index_by_id.get(parent_id) if parent_id else None
|
||||
if parent is None:
|
||||
r = parent_id or key
|
||||
break
|
||||
cur = parent
|
||||
for k in chain:
|
||||
resolved[k] = r
|
||||
return r
|
||||
|
||||
return root_of
|
||||
|
||||
|
||||
def _thread_members(index: list[dict], target: dict) -> list[dict]:
|
||||
by_id = {m["msgid"]: m for m in index}
|
||||
root_of = _make_root_resolver(by_id)
|
||||
root = root_of(target)
|
||||
members = [m for m in index if root_of(m) == root]
|
||||
return members
|
||||
|
||||
|
||||
def _fetch_bodies(s, members: list[dict]) -> list[dict]:
|
||||
out = []
|
||||
for m in members[-MAX_THREAD_MSGS:]:
|
||||
entry = dict(m)
|
||||
try:
|
||||
_, info = s.article(str(m["num"]))
|
||||
msg = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy)
|
||||
entry["body"] = _clean_quotes(_extract_body(msg))
|
||||
except Exception as e:
|
||||
entry["body"] = f"[Не удалось прочитать: {e}]"
|
||||
out.append(entry)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- инструменты
|
||||
|
||||
def scan_group(server: str, group: str, port: int = 119,
|
||||
user: str = "", password: str = "",
|
||||
window: int = DEFAULT_WINDOW) -> dict:
|
||||
"""
|
||||
НАЙТИ НОВОЕ: сравнивает группу с отметкой прошлого запуска, собирает
|
||||
затронутые треды целиком и возвращает их с контекстом.
|
||||
"""
|
||||
state = _load_state()
|
||||
key = _state_key(server, group)
|
||||
st = state.setdefault(key, {"last_seen": None, "my_ids": []})
|
||||
my_ids = set(st.get("my_ids", []))
|
||||
first_run = st["last_seen"] is None
|
||||
|
||||
try:
|
||||
with _connect(server, port, user, password) as s:
|
||||
_, count, first, last, _ = s.group(group)
|
||||
first, last = int(first), int(last)
|
||||
|
||||
if first_run:
|
||||
st["last_seen"] = max(first, last - 50)
|
||||
since = st["last_seen"]
|
||||
|
||||
if last <= since:
|
||||
_save_state(state)
|
||||
return {"status": "no_new_messages", "group": group,
|
||||
"last_seen": since, "last_num": last, "threads": []}
|
||||
|
||||
index = _fetch_index(s, last, window)
|
||||
new_msgs = [m for m in index if m["num"] > since]
|
||||
|
||||
if not new_msgs:
|
||||
st["last_seen"] = last
|
||||
_save_state(state)
|
||||
return {"status": "no_new_messages_in_window", "group": group,
|
||||
"last_num": last, "threads": []}
|
||||
|
||||
threads = {}
|
||||
for m in new_msgs:
|
||||
members = _thread_members(index, m)
|
||||
members_sorted = sorted(members, key=lambda x: x["num"])
|
||||
root = members_sorted[0]["msgid"]
|
||||
t = threads.setdefault(root, {"members": {}, "new": set()})
|
||||
for mem in members_sorted:
|
||||
t["members"][mem["msgid"]] = mem
|
||||
t["new"].add(m["msgid"])
|
||||
|
||||
result_threads = []
|
||||
for root, t in list(threads.items())[:10]:
|
||||
members = sorted(t["members"].values(), key=lambda x: x["num"])
|
||||
full = _fetch_bodies(s, members)
|
||||
refs_all = {r for m in members for r in m["refs"]}
|
||||
msg_ids = {m["msgid"] for m in members}
|
||||
reply_to_me = bool(refs_all & my_ids) or bool(msg_ids & my_ids)
|
||||
last_is_mine = members[-1]["msgid"] in my_ids
|
||||
result_threads.append({
|
||||
"root_msgid": root,
|
||||
"subject": full[0]["subject"] or full[-1]["subject"],
|
||||
"reply_to_me": reply_to_me,
|
||||
"last_is_mine": last_is_mine,
|
||||
"needs_reply": bool(t["new"] - msg_ids & my_ids) and not last_is_mine,
|
||||
"new_msg_nums": sorted(
|
||||
m["num"] for m in members if m["msgid"] in t["new"]),
|
||||
"messages": [
|
||||
{"num": m["num"], "msgid": m["msgid"],
|
||||
"from": m["from"], "date": m["date"], "body": m["body"]}
|
||||
for m in full
|
||||
],
|
||||
})
|
||||
|
||||
st["last_seen"] = last
|
||||
_save_state(state)
|
||||
return {"status": "ok", "group": group, "first_run": first_run,
|
||||
"last_num": last, "threads": result_threads}
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "group": group, "error": str(e)}
|
||||
|
||||
|
||||
def read_thread(server: str, group: str, msg_id: str, port: int = 119,
|
||||
user: str = "", password: str = "",
|
||||
window: int = DEFAULT_WINDOW) -> dict:
|
||||
"""
|
||||
ПРОЧИТАТЬ ВЕТКУ ЦЕЛИКОМ: msg_id — Message-ID (<...>) или номер статьи.
|
||||
Возвращает все сообщения треда в хронологическом порядке с телами.
|
||||
"""
|
||||
try:
|
||||
with _connect(server, port, user, password) as s:
|
||||
_, count, first, last, _ = s.group(group)
|
||||
last_num = int(last)
|
||||
|
||||
try:
|
||||
_, info = s.article(msg_id)
|
||||
probe = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy)
|
||||
target_id = probe.get("Message-ID", msg_id)
|
||||
except Exception as e:
|
||||
return {"error": f"Сообщение {msg_id} не найдено: {e}"}
|
||||
|
||||
index = _fetch_index(s, last_num, window)
|
||||
by_id = {m["msgid"]: m for m in index}
|
||||
target = by_id.get(target_id)
|
||||
if target is None:
|
||||
return {"error": f"{target_id} вне окна сканирования, увеличьте window"}
|
||||
|
||||
members = _thread_members(index, target)
|
||||
full = _fetch_bodies(s, members)
|
||||
return {"thread_root": full[0]["msgid"],
|
||||
"subject": full[0]["subject"],
|
||||
"messages": [{"num": m["num"], "msgid": m["msgid"],
|
||||
"from": m["from"], "date": m["date"],
|
||||
"body": m["body"]} for m in full]}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
def post_reply(server: str, group: str, reply_to_msg_id: str, body: str,
|
||||
sender_name: str, sender_email: str,
|
||||
port: int = 119, user: str = "", password: str = "") -> dict:
|
||||
"""
|
||||
ОТВЕТИТЬ: постит ответ на сообщение reply_to_msg_id в группу.
|
||||
Автоматически проставляет Re: в Subject, In-Reply-To и References.
|
||||
"""
|
||||
if not re.match(r"[^@\s]+@[^@\s]+", sender_email):
|
||||
return {"error": "Некорректный sender_email"}
|
||||
try:
|
||||
with _connect(server, port, user, password) as s:
|
||||
s.group(group)
|
||||
try:
|
||||
_, info = s.article(reply_to_msg_id)
|
||||
except Exception as e:
|
||||
return {"error": f"Не найдено сообщение {reply_to_msg_id}: {e}"}
|
||||
|
||||
parent = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy)
|
||||
parent_id = (parent.get("Message-ID") or reply_to_msg_id).strip()
|
||||
parent_subject = _decode(parent.get("Subject", ""))
|
||||
subject = parent_subject if re.match(r"(?i)^\s*(re|aw)\s*:", parent_subject) \
|
||||
else f"Re: {parent_subject}"
|
||||
|
||||
refs = [r for r in (parent.get("References") or "").split()
|
||||
if r.startswith("<")]
|
||||
refs.append(parent_id)
|
||||
|
||||
msg = EmailMessage()
|
||||
msg["From"] = formataddr((sender_name, sender_email))
|
||||
msg["Newsgroups"] = group
|
||||
msg["Subject"] = subject
|
||||
msg["Date"] = formatdate(localtime=True)
|
||||
msg["Message-ID"] = make_msgid(domain=sender_email.split("@")[-1])
|
||||
msg["In-Reply-To"] = parent_id
|
||||
msg["References"] = " ".join(refs)
|
||||
msg.set_content(body, charset="utf-8")
|
||||
|
||||
data = msg.as_bytes(policy=email_policy)
|
||||
resp = s.post(data.split(b"\n"))
|
||||
|
||||
state = _load_state()
|
||||
st = state.setdefault(_state_key(server, group),
|
||||
{"last_seen": None, "my_ids": []})
|
||||
st.setdefault("my_ids", []).append(msg["Message-ID"])
|
||||
st["my_ids"] = st["my_ids"][-200:]
|
||||
_save_state(state)
|
||||
|
||||
return {"status": "posted", "message_id": msg["Message-ID"],
|
||||
"subject": subject, "server_response": str(resp)}
|
||||
except Exception as e:
|
||||
return {"error": f"Ошибка отправки: {e}"}
|
||||
|
||||
|
||||
def list_groups(server: str, port: int = 119,
|
||||
user: str = "", password: str = "") -> list:
|
||||
"""Диагностика: список групп на сервере."""
|
||||
try:
|
||||
with _connect(server, port, user, password) as s:
|
||||
_, groups = s.list()
|
||||
return [g for g in groups if g]
|
||||
except Exception as e:
|
||||
return [f"Ошибка: {e}"]
|
||||
|
||||
|
||||
# Безопасная централизованная регистрация инструментов через прокси-функцию
|
||||
scan_group = register_mcp_tool(scan_group)
|
||||
read_thread = register_mcp_tool(read_thread)
|
||||
post_reply = register_mcp_tool(post_reply)
|
||||
list_groups = register_mcp_tool(list_groups)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run(transport="stdio")
|
||||
|
||||
Reference in new issue
Block a user