358 lines
19 KiB
Python
358 lines
19 KiB
Python
"""MCP-инструменты модуля wiki (stdio, FastMCP, сервер sdlc-wiki) — инструменты сессии Claude поверх HTTP API хаба
|
||
(specs/mcp/spec.md «MCP вики»). Охрана, привязка к проекту и projects/project_create — в core/mcp.py. Markdown страниц и
|
||
имена вложений приходят из Eva и считаются недоверенными: страница выдаётся текстом, имя файла сводится к basename."""
|
||
import asyncio
|
||
import re
|
||
import time
|
||
from datetime import datetime
|
||
from urllib.parse import quote
|
||
|
||
from mcp.server.fastmcp import FastMCP
|
||
|
||
from ...core import mcp as core
|
||
from ...core.client import HubError
|
||
from . import roots as wiki_roots
|
||
|
||
POLL_SECONDS = 2
|
||
ATTACHMENT_WAIT = 60
|
||
SYNC_WAIT = 300
|
||
CHUNK = 20000
|
||
CANDIDATES = 10
|
||
mcp = FastMCP("sdlc-wiki")
|
||
_NEWLINES = re.compile(r"\s*[\r\n]+\s*")
|
||
|
||
|
||
def _line(v) -> str:
|
||
"""Значение из Eva внутри строки ответа: переводы строк — пробел, иначе они подделали бы строки шапки (T9.3)."""
|
||
return _NEWLINES.sub(" ", core.value(v))
|
||
|
||
|
||
def _project() -> dict:
|
||
return next(p for p in core.client().get("/projects")["projects"] if p["label"] == core.PROJECT)
|
||
|
||
|
||
def _no_root(label) -> str:
|
||
return f"у проекта {label} нет корня вики: root_set DOC-…"
|
||
|
||
|
||
def _gate(need_sync: bool = True) -> str | None:
|
||
"""Отказ вместо запроса к вики: нет корня или (need_sync) нет ни одной синхронизации. None — можно работать."""
|
||
p = _project()
|
||
if not p.get("wiki_root"):
|
||
return _no_root(p["label"])
|
||
if need_sync and not p.get("wiki_synced_at"):
|
||
return "вики проекта ещё не синхронизирована: sync"
|
||
return None
|
||
|
||
|
||
def _walk(pages: list, code: str, depth: int) -> list:
|
||
"""Страницы поддерева code в порядке обхода до depth уровней ниже него; code не найден — None."""
|
||
start = next((i for i, p in enumerate(pages) if p["code"] == code), None)
|
||
if start is None:
|
||
return None
|
||
base = pages[start]["depth"]
|
||
out = [pages[start]]
|
||
for p in pages[start + 1:]:
|
||
if p["depth"] <= base:
|
||
break
|
||
out.append(p)
|
||
return [p for p in out if p["depth"] - base <= depth]
|
||
|
||
|
||
def _stamp(at) -> str:
|
||
"""Время из базы (UTC) → «ДД.ММ ЧЧ:ММ» в местном поясе машины MCP; пусто — «—»."""
|
||
return "—" if not at else datetime.fromisoformat(at).astimezone().strftime("%d.%m %H:%M")
|
||
|
||
|
||
def _who(a) -> str:
|
||
"""Автор правки: имя, иначе логин, иначе «—»."""
|
||
a = a or {}
|
||
return _line(a.get("name") or a.get("login") or None)
|
||
|
||
|
||
def _modified(d) -> str:
|
||
if not d.get("modified_at"):
|
||
return "изменено: история не загружена"
|
||
return f"изменено: {_stamp(d['modified_at'])} {_who(d.get('modified_by'))} (версий: {d.get('versions') or 0})"
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def tree(code: str = "", depth: int = 2) -> str:
|
||
"""Дерево вики проекта: от корня или от страницы code, depth уровней ниже неё (по умолчанию 2). Строка — отступ, код,
|
||
название, размер Markdown в символах."""
|
||
if (refusal := _gate()) is not None:
|
||
return refusal
|
||
pages = core.client().get("/wiki/tree", project=core.PROJECT)["pages"]
|
||
if not pages:
|
||
return "страниц нет"
|
||
depth = max(int(depth), 0)
|
||
if code:
|
||
shown = _walk(pages, code.strip(), depth)
|
||
if shown is None:
|
||
return f"нет страницы {code}; дерево без кода — tree"
|
||
else:
|
||
base = min(p["depth"] for p in pages)
|
||
shown = [p for p in pages if p["depth"] - base <= depth]
|
||
base = shown[0]["depth"]
|
||
return "\n".join(f"{' ' * (p['depth'] - base)}{p['code']} {_line(p.get('title'))} ({core.value(p.get('size'))} симв.)"
|
||
for p in shown)
|
||
|
||
|
||
def _candidates(pages: list, title: str) -> str:
|
||
low = title.casefold()
|
||
found = [p for p in pages if low in (p.get("title") or "").casefold()]
|
||
if not found:
|
||
return f"нет страницы «{_line(title)}»; дерево — tree, поиск по тексту — search"
|
||
lines = [f"{p['code']} {_line(p.get('title'))}" for p in found[:CANDIDATES]]
|
||
more = [f"…и ещё {len(found) - CANDIDATES}"] if len(found) > CANDIDATES else []
|
||
return "под название «%s» подходит несколько страниц, повторите page с code:\n" % _line(title) + "\n".join(lines + more)
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def page(code: str = "", title: str = "", offset: int = 0) -> str:
|
||
"""Страница вики: шапка (код, путь в дереве, адрес в Eva, дата синхронизации, вложения с id) и Markdown с символа
|
||
offset, до 20 000 символов за вызов; если текст длиннее, последняя строка — «продолжение: offset=<N>». Страница
|
||
задаётся кодом (DOC-…) или названием: точное название без учёта регистра открывает страницу, иначе будут показаны
|
||
до 10 кандидатов с кодами."""
|
||
if (refusal := _gate()) is not None:
|
||
return refusal
|
||
client = core.client()
|
||
code, title = (code or "").strip(), (title or "").strip()
|
||
if not code:
|
||
if not title:
|
||
return "нужен code или title"
|
||
pages = client.get("/wiki/tree", project=core.PROJECT)["pages"]
|
||
exact = [p for p in pages if (p.get("title") or "").casefold() == title.casefold()]
|
||
if len(exact) != 1:
|
||
return _candidates(exact or pages, title)
|
||
code = exact[0]["code"]
|
||
d = client.get(f"/wiki/pages/{quote(code, safe='')}", project=core.PROJECT)
|
||
text, offset = d.get("markdown") or "", max(int(offset), 0)
|
||
atts = d.get("attachments") or []
|
||
head = [f"{_line(d['code'])} — {_line(d.get('title'))}", f"путь: {_line(d.get('path'))}",
|
||
f"Eva: {_line(d.get('url'))}", f"синхронизировано: {_line(d.get('synced_at'))}", _modified(d)]
|
||
head += ([f"вложения ({len(atts)}):"] + [f" #{a['id']} {_line(a.get('name'))} ({_line(a.get('state'))})" for a in atts]
|
||
if atts else ["вложений нет"])
|
||
if offset >= len(text) and (offset > 0 or not text):
|
||
body = f"offset={offset} за концом текста: длина {len(text)} символов" if offset > 0 else "страница пуста"
|
||
return "\n".join(head + ["", body])
|
||
end = offset + CHUNK
|
||
out = "\n".join(head + ["", text[offset:end]])
|
||
return out + f"\nпродолжение: offset={end}" if end < len(text) else out
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def search(query: str, limit: int = 10) -> str:
|
||
"""Полнотекстовый поиск по страницам вики проекта: код, название, путь и фрагмент (limit — 1…50, по умолчанию 10)."""
|
||
if (refusal := _gate()) is not None:
|
||
return refusal
|
||
items = core.client().get("/wiki/search", project=core.PROJECT, q=query, limit=limit)["results"]
|
||
if not items:
|
||
return "ничего не найдено"
|
||
return "\n".join(f"{r['code']} {_line(r.get('title'))} — {_line(r.get('path'))}\n {_line(r.get('snippet'))}"
|
||
for r in items)
|
||
|
||
|
||
def _store(attachment_id: int, name, data: bytes):
|
||
"""Копия вложения в downloads/wiki-<id>/<basename>: имя из Eva недоверенное — берётся последний компонент пути."""
|
||
base = (name or "").replace("\\", "/").rsplit("/", 1)[-1]
|
||
return core.store_copy(f"wiki-{attachment_id}", core.safe_name(base, str(attachment_id)), data)
|
||
|
||
|
||
def fmt_wiki_changes(data: dict, now: datetime) -> str:
|
||
"""Правки по страницам: страницы — по времени последней правки (лента уже новые сверху), под страницей — её правки."""
|
||
lines = [f"правки вики с {_stamp(data['since'])} по {_stamp(data['until'])}",
|
||
f"синхронизировано: {_stamp(data.get('synced_at'))}"]
|
||
if data.get("history_at") is None:
|
||
return "\n".join(lines + ["история ещё не загружена: нужен eva-agent v2 и sync"])
|
||
events = data.get("changes") or []
|
||
if not events:
|
||
return "\n".join(lines + ["за период правок нет"])
|
||
groups: dict[str, list] = {}
|
||
for c in events:
|
||
groups.setdefault(c["page_code"], []).append(c)
|
||
for code, items in groups.items():
|
||
lines.append(f"{code} {_line(items[0].get('title'))} — {_line(items[0].get('path'))}")
|
||
for c in items:
|
||
local = datetime.fromisoformat(c["at"]).astimezone()
|
||
when = local.strftime("%H:%M") if local.date() == now.date() else local.strftime("%d.%m %H:%M")
|
||
lines.append(f" {when} {_who(c.get('author'))}: +{c['added']} / −{c['removed']} ({_line(c['version'])})")
|
||
if data.get("truncated"):
|
||
lines.append(f"показаны последние {len(events)} правок, сузьте период")
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def changes(hours: float | None = None, since: str = "", until: str = "", author: str = "") -> str:
|
||
"""Правки страниц вики проекта за период: по умолчанию — с начала сегодняшних суток; hours — за последние N часов;
|
||
since/until — ISO-8601 (без смещения — местное время); author — подстрока имени или логина без учёта регистра.
|
||
Команд не ставит: читает копию хаба."""
|
||
if hours is not None and since:
|
||
return "либо hours, либо since"
|
||
if hours is not None and hours <= 0:
|
||
return "hours — число больше 0"
|
||
if (refusal := _gate()) is not None:
|
||
return refusal
|
||
now = datetime.now().astimezone()
|
||
data = core.client().get("/wiki/changes", project=core.PROJECT, since=core.local_if_naive(core.since_of(hours, since, now)),
|
||
until=core.local_if_naive(until), author=author)
|
||
return fmt_wiki_changes(data, now)
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def diff(code: str, version: str = "", offset: int = 0) -> str:
|
||
"""Дифф правки страницы вики: шапка (код, название, путь, версия «N из M», автор, время, «+A / −R», предыдущая
|
||
версия) и unified diff с offset, до 20 000 символов за вызов; если дифф длиннее, последняя строка — «продолжение:
|
||
offset=<N>». Без version — последняя правка. Команд не ставит."""
|
||
if (refusal := _gate()) is not None:
|
||
return refusal
|
||
params = {"project": core.PROJECT, "offset": max(int(offset), 0)}
|
||
if version:
|
||
params["version"] = version.strip()
|
||
d = core.client().get(f"/wiki/pages/{quote(code.strip(), safe='')}/diff", **params)
|
||
prev = d.get("prev_version")
|
||
head = [f"{_line(d['code'])} — {_line(d.get('title'))}", f"путь: {_line(d.get('path'))}",
|
||
f"версия {d['index']} из {d['total']}: {_line(d['version'])}", f"автор: {_who(d.get('author'))}",
|
||
f"время: {_stamp(d.get('at'))}", f"+{d['added']} / −{d['removed']}",
|
||
f"предыдущая: {_line(prev) if prev else 'нет (первая версия)'}"]
|
||
out = "\n".join(head + ["", d.get("diff") or ""])
|
||
nxt = d.get("next_offset")
|
||
return out + f"\nпродолжение: offset={nxt}" if nxt is not None else out
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
async def attachment(id: int) -> str:
|
||
"""Скачать вложение страницы вики и вернуть путь к локальной копии (читайте файл по этому пути). id — номер из
|
||
ответа page. Хаб просит агента скачать файл; ожидание до 60 с, не дождался — вызовите инструмент ещё раз: вторая
|
||
команда не поставится."""
|
||
refusal = await asyncio.to_thread(_gate)
|
||
if refusal is not None:
|
||
return refusal
|
||
client = core.client()
|
||
st = await asyncio.to_thread(client.post, f"/wiki/attachments/{id}/fetch", {"author": core.AUTHOR})
|
||
deadline = time.monotonic() + ATTACHMENT_WAIT
|
||
while st.get("state") == "fetching" and time.monotonic() < deadline:
|
||
await asyncio.sleep(POLL_SECONDS)
|
||
st = await asyncio.to_thread(client.get, f"/wiki/attachments/{id}")
|
||
if st.get("state") == "fetching":
|
||
return f"скачивание ещё идёт (вложение #{id}); вызовите attachment ещё раз — вторая команда не поставится"
|
||
if st.get("state") != "ready":
|
||
return f"не удалось скачать: {_line(st.get('error'))}"
|
||
data = await asyncio.to_thread(client.get_bytes, f"/wiki/attachments/{id}/content")
|
||
path = await asyncio.to_thread(_store, id, st.get("name"), data)
|
||
return f"{path}\n{_line(st.get('name'))} ({len(data)} байт), страница {_line(st.get('page_code'))}"
|
||
|
||
|
||
def _summary(client) -> str:
|
||
s = client.get("/wiki/status", project=core.PROJECT)
|
||
lines = [f"корень: {core.value(s.get('root'))} {_line(s.get('root_title'))}; страниц: {core.value(s.get('pages'))}; "
|
||
f"последняя синхронизация: {core.value(s.get('synced_at'))}"]
|
||
if s.get("error"):
|
||
lines.append(f"ошибка синхронизации: {_line(s['error'])}")
|
||
b = s.get("batch")
|
||
lines.append(f"пакет: #{b['id']} {b['status']}" if b else "пакета в работе нет")
|
||
lines.append(f"история: {s['history_at']}" if s.get("history_at") else "история не загружена")
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
async def sync(wait: bool = False) -> str:
|
||
"""Синхронизировать вики проекта с Eva: ставит пакет и отвечает «поставлено: пакет N» или «уже идёт: пакет N».
|
||
wait=true — ждёт итога до 300 с: «ок: синхронизировано ДД.ММ ЧЧ:ММ», «ошибка: <текст>» или «не дождался за 300 с:
|
||
пакет N». Данных копии в ответе нет — их читают page, changes, status."""
|
||
refusal = await asyncio.to_thread(_gate, False)
|
||
if refusal is not None:
|
||
return refusal
|
||
client = core.client()
|
||
try:
|
||
batch_id = (await asyncio.to_thread(client.post, "/wiki/sync", {"project": core.PROJECT, "author": core.AUTHOR}))["batch_id"]
|
||
head = f"поставлено: пакет {batch_id}"
|
||
except HubError as e:
|
||
if e.error != "sync_in_progress" or not isinstance(e.detail, dict):
|
||
raise
|
||
batch_id = e.detail["batch_id"]
|
||
head = f"уже идёт: пакет {batch_id}"
|
||
if not wait:
|
||
return head
|
||
deadline = time.monotonic() + SYNC_WAIT
|
||
while True:
|
||
b = await asyncio.to_thread(client.get, f"/batches/{batch_id}")
|
||
if b["status"] in ("done", "cancelled") or time.monotonic() >= deadline:
|
||
break
|
||
await asyncio.sleep(POLL_SECONDS)
|
||
if b["status"] not in ("done", "cancelled"):
|
||
return f"не дождался за {SYNC_WAIT:g} с: пакет {batch_id}"
|
||
if b["status"] == "cancelled":
|
||
return f"ошибка: пакет {batch_id} отменён"
|
||
failed = next((c for c in b.get("commands") or [] if c.get("status") == "failed"), None)
|
||
if failed is not None:
|
||
return f"ошибка: {_line(failed.get('error'))}"
|
||
st = await asyncio.to_thread(client.get, "/wiki/status", project=core.PROJECT)
|
||
return f"ок: синхронизировано {_stamp(st.get('synced_at'))}"
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def status() -> str:
|
||
"""Состояние вики проекта: корень, число страниц, последняя синхронизация, ошибка, пакет в работе; затем агент wiki
|
||
(онлайн/офлайн, версия, публикация, обновление, запускатель)."""
|
||
client = core.client()
|
||
p = _project()
|
||
head = _summary(client) if p.get("wiki_root") else _no_root(p["label"])
|
||
return head + "\n" + core.fmt_agent(client.get("/agent", module="wiki"))
|
||
|
||
|
||
@mcp.tool()
|
||
@core.guard
|
||
@core.scoped(True)
|
||
def root_set(root: str) -> str:
|
||
"""Задать корень вики проекта: код DOC-… или ссылка на документ Eva. Хаб ставит синхронизацию нового корня; прежняя
|
||
копия остаётся до её успеха."""
|
||
try:
|
||
code = wiki_roots.parse_root(root)
|
||
except ValueError as e:
|
||
return f"корень не задан: {e}; пример: root_set DOC-123"
|
||
if code is None:
|
||
return "корень не задан: укажите код DOC-… или ссылку на документ Eva"
|
||
out = core.client().patch(f"/projects/{quote(core.PROJECT, safe='')}", {"wiki_root": code, "author": core.AUTHOR})
|
||
ids = out.get("batch_ids") or []
|
||
return (f"корень вики проекта {core.PROJECT}: {code}; синхронизация поставлена: пакет #{ids[0]}" if ids
|
||
else f"корень вики проекта {core.PROJECT}: {code}; пакета синхронизации не поставлено (корень не изменился)")
|
||
|
||
|
||
@core.guard
|
||
def project_create(label: str, name: str, wiki_root: str = "") -> str:
|
||
"""Завести проект (метка, название, необязательно корень вики: код DOC-… или ссылка Eva). С корнем хаб ставит
|
||
синхронизацию; в ответе — фрагмент .mcp.json для репозитория проекта с сервером sdlc-wiki."""
|
||
try:
|
||
code = wiki_roots.parse_root(wiki_root)
|
||
except ValueError as e:
|
||
return f"проект не заведён: {e}"
|
||
return core.create_project(label, name, {} if code is None else {"wiki_root": code}, "wiki")
|
||
|
||
|
||
PROJECT_COLUMNS = [("корень вики", lambda p: core.value(p.get("wiki_root"))),
|
||
("страниц", lambda p: core.value(p.get("wiki_pages"))),
|
||
("синхронизация", lambda p: core.value(p.get("wiki_synced_at")))]
|
||
|
||
core.register_core_tools(
|
||
mcp,
|
||
projects_doc="Проекты хаба: метка, название, корень вики, страниц, последняя синхронизация; текущий проект сессии отмечен.",
|
||
columns=PROJECT_COLUMNS,
|
||
project_create=project_create)
|