从零写一个 MCP stdio Server：用 Python 校验博客文章格式

import json
import os
import re
import sys
from pathlib import Path

ROOT = Path(os.environ.get("ARTICLES_DIR", "./articles")).resolve()
PROTOCOL_FALLBACK = "2025-06-18"

def log(*args):
    print(*args, file=sys.stderr, flush=True)

def reply(mid, result):
    return {"jsonrpc": "2.0", "id": mid, "result": result}

TOOLS = [
    {
        "name": "list_articles",
        "description": "列出文章目录下的 Markdown 文件及其标题",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "check_article",
        "description": "检查标题长度、tags 数量、正文汉字数、代码块、表格和文末引用块",
        "inputSchema": {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "文章文件名，例如 2026-11-08-xxx.md"}
            },
            "required": ["name"],
        },
    },
]

def handle(msg):
    method = msg.get("method")
    mid = msg.get("id")
    params = msg.get("params") or {}
    if mid is None:          # 通知没有 id，直接忽略
        return None
    if method == "initialize":
        return reply(mid, {
            "protocolVersion": params.get("protocolVersion", PROTOCOL_FALLBACK),
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "article-mcp", "version": "0.1.0"},
        })
    if method == "ping":
        return reply(mid, {})
    if method == "tools/list":
        return reply(mid, {"tools": TOOLS})
    if method == "tools/call":
        return reply(mid, call_tool(params.get("name"), params.get("arguments") or {}))
    return {"jsonrpc": "2.0", "id": mid,
            "error": {"code": -32601, "message": f"未知方法 {method}"}}

def main():
    log(f"article-mcp 已启动，只读目录：{ROOT}")
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            out = handle(json.loads(line))
        except Exception as exc:
            out = {"jsonrpc": "2.0", "id": None,
                   "error": {"code": -32700, "message": str(exc)}}
        if out:
            print(json.dumps(out, ensure_ascii=False), flush=True)

CJK = re.compile(r"[\u4e00-\u9fff]")
FOOTER = "> 个人博客：https://www.sakura.work.gd/ ｜ GitHub：https://github.com/sakurajh"

def parse_front_matter(text):
    if not text.startswith("---"):
        return {}, text
    parts = text.split("---", 2)
    if len(parts) < 3:
        return {}, text
    meta = {}
    for line in parts[1].strip().splitlines():
        key, sep, value = line.partition(":")
        if not sep:
            continue
        value = value.strip()
        if value.startswith("[") and value.endswith("]"):
            value = [v.strip().strip("\"'") for v in value[1:-1].split(",") if v.strip()]
        else:
            value = value.strip("\"'")
        meta[key.strip()] = value
    return meta, parts[2]

def safe_path(name):
    path = (ROOT / name).resolve()
    if ROOT not in path.parents:
        raise ValueError("只允许访问 ARTICLES_DIR 下的文件")
    return path

def check_article(name):
    path = safe_path(name)
    meta, body = parse_front_matter(path.read_text(encoding="utf-8"))
    problems = []
    title = meta.get("title", "")
    if not 20 <= len(title) <= 45:
        problems.append(f"标题 {len(title)} 个字符，应在 20–45 之间")
    tags = meta.get("tags") or []
    if isinstance(tags, str):
        tags = [t for t in re.split(r"[,，\s]+", tags) if t]
    if not 4 <= len(tags) <= 6:
        problems.append(f"tags 有 {len(tags)} 个，应为 4–6 个")
    for key in ("summary", "slug", "category"):
        if not meta.get(key):
            problems.append(f"front matter 缺少 {key}")
    cjk = len(CJK.findall(body))
    if cjk < 2000:
        problems.append(f"正文汉字 {cjk} 个，少于 2000")
    if body.count(chr(96) * 3) < 2:
        problems.append("没有成对的代码块")
    if not re.search(r"^\|[\s:|-]+\|\s*$", body, re.M):
        problems.append("没有表格分隔行")
    if FOOTER not in body:
        problems.append("缺少文末引用块")
    report = f"{path.name}：正文 {cjk} 个汉字"
    return report + "\n" + ("\n".join("- " + p for p in problems) if problems else "检查通过")

def call_tool(name, args):
    try:
        if name == "list_articles":
            text = "\n".join(
                f"{p.name}\t{parse_front_matter(p.read_text(encoding='utf-8'))[0].get('title', '(无标题)')}"
                for p in sorted(ROOT.glob("*.md"), reverse=True)
            ) or "目录下没有 Markdown 文件"
        elif name == "check_article":
            text = check_article(args.get("name", ""))
        else:
            raise ValueError(f"未知工具 {name}")
        return {"content": [{"type": "text", "text": text}], "isError": False}
    except Exception as exc:
        return {"content": [{"type": "text", "text": str(exc)}], "isError": True}

if __name__ == "__main__":
    main()