feat(community): add functional packages and phase three delivery plan

2026-09-06 01:19:29 +08:00
parent 99a92e9eb1
commit 9497519e8b
19 changed files with 675 additions and 2 deletions
@@ -0,0 +1,130 @@
+"""Markdown checks over MCP stdio; Python standard library only, no I/O tools."""
+from __future__ import annotations
+
+import json
+import re
+import sys
+
+VERSION = '1.0.0'
+MAX_TEXT = 100_000
+MAX_ITEMS = 200
+
+
+def inspect_markdown(text: str) -> dict:
+    if not isinstance(text, str) or len(text) > MAX_TEXT:
+        raise ValueError('text 必须是字符串，最多 100000 个字符。')
+    lines = text.splitlines()
+    headings, tasks, issues = [], [], []
+    previous_level = 0
+    titles = set()
+    fence = None
+    frontmatter_end = -1
+    if lines and lines[0].lstrip('\ufeff') == '---':
+        frontmatter_end = next((i for i in range(1, len(lines)) if lines[i] in ('---', '...')), -1)
+    for index, line in enumerate(lines):
+        number = index + 1
+        if index <= frontmatter_end:
+            continue
+        marker = re.match(r'^ {0,3}(`{3,}|~{3,})(.*)$', line)
+        if fence:
+            if marker and marker[1][0] == fence[0] and len(marker[1]) >= fence[1] and not marker[2].strip():
+                fence = None
+            continue
+        if marker and not (marker[1][0] == '`' and '`' in marker[2]):
+            fence = (marker[1][0], len(marker[1]), number)
+            continue
+        # Indented code and blockquotes are excluded from these line-based checks.
+        if line.startswith(('    ', '\t', '>')):
+            continue
+        heading = re.match(r'^ {0,3}(#{1,6})(?:\s+(.*)|$)', line)
+        level, title = 0, ''
+        if heading:
+            level = len(heading[1])
+            title = re.sub(r'\s+#+\s*$', '', heading[2] or '').strip()
+        elif index + 1 < len(lines) and line.strip() and re.fullmatch(r' {0,3}(=+|-+)\s*', lines[index + 1]) and not re.match(r'^\s*(?:[-*+]\s|\d+[.)]\s|[-=]+\s*$)', line):
+            level = 1 if lines[index + 1].lstrip().startswith('=') else 2
+            title = line.strip()
+        if level:
+            headings.append({'line': number, 'level': level, 'title': title[:300]})
+            if previous_level and level > previous_level + 1:
+                issues.append({'line': number, 'code': 'heading_jump', 'message': f'标题从 H{previous_level} 跳到 H{level}。'})
+            if title.casefold() in titles:
+                issues.append({'line': number, 'code': 'duplicate_heading', 'message': '存在同名标题，请确认是否需要区分。'})
+            if not title:
+                issues.append({'line': number, 'code': 'empty_heading', 'message': '标题内容为空。'})
+            titles.add(title.casefold())
+            previous_level = level
+        task = re.match(r'^ {0,3}(?:[-*+]|\d+[.)])\s+\[([ xX])\]\s+(.*)$', line)
+        if task:
+            tasks.append({'line': number, 'done': task[1].lower() == 'x', 'text': task[2][:300]})
+    if fence:
+        issues.append({'line': fence[2], 'code': 'unclosed_fence', 'message': '代码围栏没有闭合。'})
+    return {
+        'summary': {'lines': len(lines), 'characters': len(text), 'headings': len(headings),
+                    'tasks': len(tasks), 'open_tasks': sum(not item['done'] for item in tasks), 'issues': len(issues)},
+        'headings': headings[:MAX_ITEMS], 'tasks': tasks[:MAX_ITEMS], 'issues': issues[:MAX_ITEMS],
+        'truncated': any(len(items) > MAX_ITEMS for items in (headings, tasks, issues)),
+        'method': 'line-based Markdown checks; line numbers refer to the supplied text',
+    }
+
+
+TOOLS = [
+    {'name': 'inspect_markdown', 'description': '本地检查 Markdown，返回标题、待办事项、格式问题及 1 起始行号。不会读取或修改文件。',
+     'inputSchema': {'type': 'object', 'properties': {'text': {'type': 'string', 'maxLength': MAX_TEXT}}, 'required': ['text'], 'additionalProperties': False}},
+    {'name': 'selection_report', 'description': 'NotesAgent 当前选区检查命令。',
+     'inputSchema': {'type': 'object', 'properties': {'_notesagent': {'type': 'object'}}, 'required': ['_notesagent'], 'additionalProperties': False}},
+]
+
+
+def call_tool(name: str, arguments: dict) -> dict:
+    if name == 'inspect_markdown':
+        result = inspect_markdown(arguments.get('text'))
+    elif name == 'selection_report':
+        envelope = arguments.get('_notesagent', {})
+        if not isinstance(envelope, dict) or not isinstance(envelope.get('context', {}), dict):
+            raise ValueError('命令上下文无效。')
+        report = inspect_markdown(envelope.get('context', {}).get('selection', ''))
+        summary = report['summary']
+        details = '；'.join(f"第 {item['line']} 行：{item['message']}" for item in report['issues'][:3])
+        result = {'type': 'notification', 'payload': {'level': 'info', 'message':
+            f"Markdown 检查：{summary['lines']} 行，{summary['headings']} 个标题，{summary['open_tasks']} 项未完成任务，{summary['issues']} 项提示。" + details}}
+    else:
+        raise ValueError('未知工具。')
+    return {'content': [{'type': 'text', 'text': json.dumps(result, ensure_ascii=False)}], 'structuredContent': result, 'isError': False}
+
+
+def main() -> None:
+    sys.stdin.reconfigure(encoding='utf-8')
+    sys.stdout.reconfigure(encoding='utf-8')
+    for raw in sys.stdin:
+        request_id = None
+        try:
+            message = json.loads(raw)
+            if not isinstance(message, dict):
+                raise ValueError('请求必须为对象。')
+            request_id = message.get('id')
+            if request_id is None:
+                continue
+            method, params = message.get('method'), message.get('params') or {}
+            if method == 'initialize':
+                result = {'protocolVersion': params.get('protocolVersion'), 'capabilities': {'tools': {'listChanged': False}},
+                          'serverInfo': {'name': 'markdown-workbench', 'version': VERSION}}
+            elif method == 'ping':
+                result = {}
+            elif method == 'tools/list':
+                result = {'tools': TOOLS}
+            elif method == 'tools/call':
+                try:
+                    result = call_tool(params.get('name'), params.get('arguments') or {})
+                except (ValueError, TypeError, AttributeError) as error:
+                    result = {'content': [{'type': 'text', 'text': str(error)}], 'isError': True}
+            else:
+                raise ValueError('不支持的方法。')
+            response = {'jsonrpc': '2.0', 'id': request_id, 'result': result}
+        except (ValueError, TypeError, AttributeError):
+            response = {'jsonrpc': '2.0', 'id': request_id, 'error': {'code': -32600, 'message': 'Invalid request'}}
+        print(json.dumps(response, ensure_ascii=False, separators=(',', ':')), flush=True)
+
+
+if __name__ == '__main__':
+    main()