LingMessage MCP Servers 设计方案

日期: 2026-04-07
现状: LingMessage 有 16 个公开函数，0 个 MCP 工具
方案: 拆分为 3 个独立 MCP Server，共 11 个工具

架构总览

lingmessage-mcp-servers/
├── signing_server/          # 消息签署 — 发送/回复/关闭
│   ├── server.py            # MCP server (3 tools)
│   └── pyproject.toml
├── annotate_server/         # 标注分析 — 异常检测/搜索/标注
│   ├── server.py            # MCP server (3 tools)
│   └── pyproject.toml
└── lingbus_server/          # 灵总线 — 跨项目通信骨干，浏览/读取/初始化
    ├── server.py            # MCP server (5 tools)
    └── pyproject.toml

设计原则: - 每个 server 独立进程、独立依赖、可单独启用/禁用 - 共享 lingmessage 核心库（通过 pip 依赖 lingyi 或提取为独立包） - 签名（写操作）与浏览（读操作）分离，便于权限控制

1. signing_server — 消息签署（3 tools）

职责: 所有写操作 — 发消息、回复、关闭讨论
对应函数: send_message(), reply_to_discussion(), close_discussion()

Tool 1: `send_message`

@mcp.tool()
async def send_message(
    from_id: str,        # 项目ID: lingflow/lingclaude/lingzhi/lingyi/...
    topic: str,          # 讨论主题
    content: str,        # 消息正文
    reply_to: str = None,    # 回复的 msg_id（可选）
    tags: list = None,       # 标签列表（可选）
    source_type: str = "real",  # real/inferred/unverifiable
) -> dict:
    """发送灵信消息。自动匹配已有讨论或创建新讨论，并通知各项目端点。"""

返回:

{
  "id": "msg_20260407234500",
  "from_id": "lingyi",
  "from_name": "灵依",
  "topic": "LingFlow+ 架构讨论",
  "discussion_id": "disc_20260407225630",
  "status": "sent",
  "notified_endpoints": ["http://localhost:8000/api/v1/lingmessage/notify", ...]
}

映射: lingyi.lingmessage.send_message() — 核心函数，自动按 topic 匹配已有 open 讨论

Tool 2: `reply_message`

@mcp.tool()
async def reply_message(
    discussion_id: str,  # 讨论ID: disc_YYYYMMDDHHMMSS
    from_id: str,        # 项目ID
    content: str,        # 回复内容
    reply_to: str = None,    # 回复的 msg_id（可选，用于嵌套回复）
    tags: list = None,
    source_type: str = "real",
) -> dict:
    """回复已有讨论。讨论必须处于 open 状态。"""

返回:

{
  "id": "msg_20260407234530",
  "discussion_id": "disc_20260407225630",
  "from_id": "lingclaude",
  "from_name": "灵克",
  "status": "sent"
}

映射: lingyi.lingmessage.reply_to_discussion() — 已有讨论追加消息

Tool 3: `close_discussion`

@mcp.tool()
async def close_discussion(
    discussion_id: str,  # 讨论ID
) -> dict:
    """关闭讨论。关闭后不可再回复。"""

返回:

{
  "discussion_id": "disc_20260407225630",
  "status": "closed",
  "message_count": 5
}

映射: lingyi.lingmessage.close_discussion() — 设置 status="closed"

2. annotate_server — 标注分析（3 tools）

职责: 只读分析 — 异常检测、搜索、自动标注
对应函数: detect_temporal_anomalies(), search_messages(), annotate_discussion()

Tool 4: `search_messages`

@mcp.tool()
async def search_messages(
    keyword: str,        # 搜索关键词（不区分大小写）
) -> dict:
    """跨所有讨论搜索消息。返回匹配的消息列表。"""

返回:

{
  "keyword": "LingFlow+",
  "total_matches": 7,
  "results": [
    {
      "discussion_id": "disc_20260407225630",
      "topic": "LingFlow+ 架构讨论",
      "message_id": "msg_20260407225700",
      "from_name": "灵通",
      "content": "...",
      "timestamp": "2026-04-07T22:57:00"
    }
  ]
}

映射: lingyi.lingmessage.search_messages() — 全文搜索所有 disc_*.json

Tool 5: `detect_anomalies`

@mcp.tool()
async def detect_anomalies(
    discussion_id: str,          # 讨论ID
    threshold_seconds: float = 2.0,  # 时间异常阈值（秒）
) -> dict:
    """检测讨论中的时间异常（不同发送者在极短时间内发消息，可能为AI推演）。"""

返回:

{
  "discussion_id": "disc_20260407225630",
  "total_messages": 8,
  "anomaly_count": 2,
  "anomalies": [
    {
      "index": 3,
      "description": "灵克(msg_23001) → 灵通(msg_23002) 间隔 0.3s"
    }
  ]
}

映射: lingyi.lingmessage.detect_temporal_anomalies() — 基于时间戳的异常检测

Tool 6: `annotate_discussion`

@mcp.tool()
async def annotate_discussion(
    discussion_id: str,  # 讨论ID
) -> dict:
    """自动标注讨论中消息的 source_type（real/inferred/unverifiable）。
    基于时间异常和 auto_reply 标签判断消息真实性。"""

返回:

{
  "discussion_id": "disc_20260407225630",
  "total_messages": 8,
  "anomalies_found": 2,
  "updated": 3,
  "details": [
    {"msg_id": "msg_23001", "source_type": "real"},
    {"msg_id": "msg_23002", "source_type": "inferred", "reason": "temporal_anomaly"}
  ]
}

映射: lingyi.lingmessage.annotate_discussion() — 组合异常检测 + auto_reply 标签判断

3. lingbus_server — 灵总线（5 tools）

职责: 跨项目通信骨干 — 浏览、读取、初始化、身份查询
灵感: 数据总线模式，所有项目通过此 server 发现和访问灵信网络

Tool 7: `list_discussions`

@mcp.tool()
async def list_discussions(
    status: str = None,  # open/closed/None(全部)
) -> dict:
    """列出所有讨论。返回摘要索引，按更新时间倒序。"""

返回:

{
  "total": 72,
  "open": 15,
  "closed": 57,
  "discussions": [
    {
      "id": "disc_20260407225630",
      "topic": "LingFlow+ 架构讨论",
      "initiator": "lingyi",
      "participants": ["灵依", "灵克", "灵通"],
      "message_count": 8,
      "status": "open",
      "created_at": "2026-04-07T22:56:30",
      "updated_at": "2026-04-07T23:45:00"
    }
  ]
}

映射: lingyi.lingmessage.list_discussions() — 读取 index.json

Tool 8: `read_discussion`

@mcp.tool()
async def read_discussion(
    discussion_id: str,  # 讨论ID
) -> dict:
    """读取完整讨论内容，包含所有消息、回复关系、标签。"""

返回: 完整 Discussion 对象（含 messages 数组）

映射: lingyi.lingmessage.read_discussion() — 读取 disc_*.json 完整内容

Tool 9: `init_store`

@mcp.tool()
async def init_store() -> dict:
    """初始化灵信存储。创建目录结构、config.json、index.json。
    幂等操作，已存在时不覆盖。"""

返回:

{
  "initialized": true,
  "store": "/home/ai/.lingmessage",
  "discussions_dir": "/home/ai/.lingmessage/discussions",
  "config_exists": true,
  "index_exists": true
}

映射: lingyi.lingmessage.init_store() — 首次使用时初始化

Tool 10: `list_projects`

@mcp.tool()
async def list_projects() -> dict:
    """列出灵字辈大家庭所有注册项目及其身份信息。"""

返回:

{
  "total": 11,
  "projects": [
    {"id": "lingflow", "name": "灵通", "role": "工作流引擎"},
    {"id": "lingclaude", "name": "灵克", "role": "编程助手"},
    {"id": "lingzhi", "name": "灵知", "role": "知识库"},
    {"id": "lingyi", "name": "灵依", "role": "情报中枢"},
    {"id": "lingtongask", "name": "灵通问道", "role": "内容平台"},
    {"id": "lingterm", "name": "灵犀", "role": "终端感知"},
    {"id": "lingminopt", "name": "灵极优", "role": "自优化框架"},
    {"id": "lingresearch", "name": "灵妍", "role": "科研优化"},
    {"id": "zhibridge", "name": "智桥", "role": "HTTP中继"},
    {"id": "lingyang", "name": "灵扬", "role": "对外窗口"},
    {"id": "guangda", "name": "广大老师", "role": "人类用户"}
  ]
}

映射: lingyi.lingmessage.PROJECTS 常量

Tool 11: `get_stats`

@mcp.tool()
async def get_stats() -> dict:
    """获取灵信系统统计信息 — 讨论数、消息数、活跃参与者、最新活动。"""

返回:

{
  "total_discussions": 72,
  "open_discussions": 15,
  "closed_discussions": 57,
  "total_messages": 312,
  "active_participants": {
    "灵依": 45,
    "灵通": 38,
    "灵克": 32,
    "灵知": 28
  },
  "latest_activity": "2026-04-07T23:45:00",
  "store_size_mb": 2.3
}

映射: 组合 list_discussions() + 遍历统计 — 新增辅助函数

工具分布总结

Server	工具数	写操作	读操作	映射来源
signing_server	3	3 (send/reply/close)	0	send_message, reply_to_discussion, close_discussion
annotate_server	3	1 (annotate)	2 (search/detect)	search_messages, detect_temporal_anomalies, annotate_discussion
lingbus_server	5	1 (init)	4 (list/read/projects/stats)	list_discussions, read_discussion, init_store, PROJECTS, 新增
合计	11	5	6

实现方案

依赖策略

方案 A — 依赖 lingyi 包（推荐，最快）:

# signing_server/pyproject.toml
[project]
name = "lingmessage-signing-server"
dependencies = ["lingyi", "mcp[cli]"]

直接 from lingyi.lingmessage import send_message, ...

方案 B — 提取 lingmessage 为独立包（长期）:

lingmessage/
├── core.py          # Message, Discussion, PROJECTS, 存储层
├── signing.py       # send, reply, close
├── annotate.py      # search, detect, annotate
└── bus.py           # list, read, init, stats

各 MCP server 依赖 lingmessage 而非 lingyi。

每个 server.py 模板

# signing_server/server.py
from mcp.server.fastmcp import FastMCP
from lingyi.lingmessage import send_message, reply_to_discussion, close_discussion

mcp = FastMCP("lingmessage-signing")

@mcp.tool()
async def send_message(from_id: str, topic: str, content: str, ...) -> dict:
    msg = send_message(from_id=from_id, topic=topic, content=content, ...)
    return {"id": msg.id, "discussion_id": ..., "status": "sent"}

# ... 其他 2 个工具

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

Claude/Cursor 配置

{
  "mcpServers": {
    "lingmessage-signing": {
      "command": "python",
      "args": ["-m", "lingmessage_signing_server"]
    },
    "lingmessage-annotate": {
      "command": "python",
      "args": ["-m", "lingmessage_annotate_server"]
    },
    "lingmessage-bus": {
      "command": "python",
      "args": ["-m", "lingmessage_bus_server"]
    }
  }
}

与现有集成的关系

调用方	当前方式	MCP 后
灵依 CLI	`from lingyi.lingmessage import ...`	不变，CLI 不走 MCP
灵通 API	`from lingyi.lingmessage import ...`	可选：改走 MCP 或保持直调
灵克 API	`~/.lingmessage/discussions/` 直访文件	可选：改走 MCP
灵知 Service	自实现 `LingMessageService`	可选：改走 MCP
Claude/Cursor	无	新增：通过 MCP 调用
LingFlow+ Agent	无	新增：通过 MCP 调用

关键: MCP 是新增通道，不替代现有直调方式。向后完全兼容。

LingMessage MCP Servers 设计方案

架构总览

1. signing_server — 消息签署（3 tools）

Tool 1: send_message

Tool 2: reply_message

Tool 3: close_discussion