Синтакс-помощник 1С как MCP-инструмент: индекс из справки платформы и сервер на Python SDK

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

"""Строит поисковую базу SQLite (FTS5) из справки синтакс-помощника платформы 1С.

Использование:
    python hbk_index.py "C:\\Program Files (x86)\\1cv8\\8.3.27.2130\\bin\\shcntx_ru.hbk" syntax.db

Файл shcntx_ru.hbk - контейнер платформы: заголовок, таблица документов, среди них FileStorage,
внутри которого обычный ZIP со страницами справки (objects/.../methods/*.html). Читаем без внешних библиотек.
Индексируются методы, свойства и события; для каждой страницы берутся заголовок, синтаксис, параметры,
возвращаемое значение, описание, версия и доступность. Текст справки остаётся в вашей базе на вашей машине.
"""
import html
import io
import re
import sqlite3
import struct
import sys
import zipfile

INF = 0x7FFFFFFF


def read_doc(f, addr):
    """Документ контейнера: заголовок '\\r\\n%08x %08x %08x \\r\\n' (размер, размер страницы, следующая страница) и данные."""
    f.seek(addr)
    h = f.read(31)
    size, page, nxt = int(h[2:10], 16), int(h[11:19], 16), int(h[20:28], 16)
    out, need = b"", size
    take = min(page, need)
    out += f.read(take)
    need -= take
    while need > 0 and nxt != INF:
        f.seek(nxt)
        h = f.read(31)
        page, nxt = int(h[11:19], 16), int(h[20:28], 16)
        take = min(page, need)
        out += f.read(take)
        need -= take
    return out


def open_storage(path):
    f = open(path, "rb")
    toc = read_doc(f, 16)  # таблица документов: тройки int32 (адрес заголовка, адрес тела, 0x7fffffff)
    docs = {}
    for i in range(0, len(toc) - 11, 12):
        head_addr, body_addr, _ = struct.unpack("<iii", toc[i:i + 12])
        name = read_doc(f, head_addr)[20:].decode("utf-16le", "replace").rstrip("\x00")
        docs[name] = body_addr
    return zipfile.ZipFile(io.BytesIO(read_doc(f, docs["FileStorage"])))


CHAPTER = re.compile(r'<p class="V8SH_chapter">(.*?)</p>', re.S)


def text(fragment):
    fragment = re.sub(r"<br\s*/?>|</p>|</div>|</li>|(?=<div)", "\n", fragment)
    fragment = re.sub(r"<[^>]+>", "", fragment)
    fragment = html.unescape(fragment).replace("\xa0", " ")
    return re.sub(r"\n\s*\n+", "\n", fragment).strip()


def parse(page):
    m = re.search(r'<h1 class="V8SH_pagetitle">(.*?)</h1>', page, re.S)
    if not m:
        return None
    title = text(m.group(1))
    parts = CHAPTER.split(page)
    chapters = {text(parts[i]).rstrip(":"): text(parts[i + 1]) for i in range(1, len(parts) - 1, 2)}
    since = re.search(r'<p class="V8SH_versionInfo">(.*?)</p>', page, re.S)
    return {
        "title": title,
        "syntax": chapters.get("Синтаксис", ""),
        "params": chapters.get("Параметры", ""),
        "returns": chapters.get("Возвращаемое значение", ""),
        "description": chapters.get("Описание", ""),
        "availability": chapters.get("Доступность", ""),
        "note": chapters.get("Примечание", ""),
        "since": text(since.group(1)) if since else "",
    }


def build(hbk, db_path):
    z = open_storage(hbk)
    db = sqlite3.connect(db_path)
    db.executescript("""
        DROP TABLE IF EXISTS entries; DROP TABLE IF EXISTS entries_fts;
        CREATE TABLE entries(id INTEGER PRIMARY KEY, kind TEXT, name_lc TEXT, title TEXT, syntax TEXT, params TEXT,
                             returns TEXT, description TEXT, availability TEXT, note TEXT, since TEXT, path TEXT);
        CREATE INDEX entries_name ON entries(name_lc);
        CREATE VIRTUAL TABLE entries_fts USING fts5(title, description, tokenize='unicode61');
    """)
    n = 0
    for name in z.namelist():
        m = re.search(r"/(methods|properties|events)/(?:[^/]+/)*[^/]+\.html$", name)
        if not m:
            continue
        e = parse(z.read(name).decode("utf-8-sig", "replace"))
        if not e:
            continue
        short = e["title"].split(" (")[0].rsplit(".", 1)[-1].lower()  # имя без типа-владельца, регистр приведён в Python: lower() в SQLite кириллицу не понимает
        cur = db.execute("INSERT INTO entries(kind,name_lc,title,syntax,params,returns,description,availability,note,since,path) "
                         "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
                         (m.group(1), short, e["title"], e["syntax"], e["params"], e["returns"], e["description"],
                          e["availability"], e["note"], e["since"], name))
        db.execute("INSERT INTO entries_fts(rowid,title,description) VALUES(?,?,?)", (cur.lastrowid, e["title"], e["description"]))
        n += 1
    db.commit()
    return n


if __name__ == "__main__":
    print("записей:", build(sys.argv[1], sys.argv[2]))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

"""MCP-сервер поиска по справке синтакс-помощника 1С (данные - syntax.db из hbk_index.py).

Запуск:  python syntax_mcp.py syntax.db          (stdio; клиент запускает процесс сам)
Зависимость: mcp>=2 (в версии 2 класс FastMCP переименован в MCPServer).
"""
import re
import sqlite3
import sys

from mcp.server.mcpserver import MCPServer

DB = sqlite3.connect(sys.argv[1] if len(sys.argv) > 1 else "syntax.db", check_same_thread=False)
DB.row_factory = sqlite3.Row
mcp = MCPServer("1c-syntax-helper")


def _fts_query(text: str) -> str:
    words = re.findall(r"\w+", text, re.U)
    return " ".join(f'"{w}"*' for w in words)


@mcp.tool()
def find_syntax(query: str, limit: int = 5) -> str:
    """Ищет метод, свойство или событие платформы 1С в справке синтакс-помощника.

    query - имя или часть имени на русском или английском, например «СтрНайти» или «ПолучитьСсылку».
    Возвращает синтаксис, параметры, версию и доступность. Пустой ответ означает, что такого имени в справке нет.
    """
    q = _fts_query(query)
    if not q:
        return "Пустой запрос."
    limit = max(1, min(limit, 20))
    # сначала точное совпадение имени, затем поиск по началу слов в заголовке
    rows = DB.execute("SELECT * FROM entries WHERE name_lc = ? LIMIT ?", (query.strip().lower(), limit)).fetchall()
    if len(rows) < limit:
        seen = {r["id"] for r in rows}
        rows += [r for r in DB.execute(
            "SELECT e.* FROM entries_fts JOIN entries e ON e.id = entries_fts.rowid "
            "WHERE entries_fts MATCH ? ORDER BY rank LIMIT ?", (f"title : ({q})", limit)).fetchall()
            if r["id"] not in seen][: limit - len(rows)]
    if not rows:
        return f"В справке синтакс-помощника нет записей по запросу «{query}»."
    out = []
    for r in rows:
        out.append(
            f"{r['title']}\nСинтаксис: {r['syntax']}\n{r['since']}\nДоступность: {r['availability']}\n"
            f"Параметры:\n{r['params']}\nВозвращает: {r['returns']}\nОписание: {r['description'][:600]}"
        )
    return "\n\n---\n\n".join(out)


if __name__ == "__main__":
    mcp.run(transport="stdio")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

"""Проверка сервера клиентом MCP SDK: python test_client.py syntax.db"""
import asyncio
import sys

from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

QUERIES = ["ПолучитьСсылку", "СтрНайти", "НачатьТранзакциюЗаписи", "НачатьТранзакцию"]


async def main(db):
    params = StdioServerParameters(command=sys.executable, args=[__file__.replace("test_client", "syntax_mcp"), db])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("инструменты:", [t.name for t in tools.tools])
            for q in QUERIES:
                res = await session.call_tool("find_syntax", {"query": q, "limit": 2})
                text = res.content[0].text
                print("=" * 70, "\nЗапрос:", q, "| символов в ответе:", len(text))
                print(text[:700])


asyncio.run(main(sys.argv[1]))

1
2
3
4
5
6
7
8
9
10

"""Проверка имён по базе справки: python name_check.py syntax.db Имя1 Имя2 ...
Печатает, есть ли такое имя метода, свойства или события и у каких типов."""
import sqlite3
import sys

db = sqlite3.connect(sys.argv[1])
for name in sys.argv[2:]:
    rows = db.execute("SELECT title FROM entries WHERE name_lc = ? LIMIT 3", (name.lower(),)).fetchall()
    total = db.execute("SELECT count(*) FROM entries WHERE name_lc = ?", (name.lower(),)).fetchone()[0]
    print(f"{name}: {'нет в справке' if not total else f'найдено {total}, например ' + rows[0][0]}")

1
2
3
4
5
6
7
8

{
  "mcpServers": {
    "1c-syntax-helper": {
      "command": "python",
      "args": ["C:/tools/syntax-mcp/syntax_mcp.py", "C:/tools/syntax-mcp/syntax.db"]
    }
  }
}