通知发布订阅服务器

Notify-MCP 是一个基于发布-订阅架构的多平台AI协作通知服务器,支持团队在不同AI助手间共享通知、决策和状态更新。

  • by osick
  • HTML
  • 本地部署
  • 更新于
在 GitHub 查看 osick/notify-mcp

使用说明

通知-MCP(或“向MCP通知”,具体翻译可能根据上下文调整,MCP可能代表某个特定的系统、部门或个人)

一个支持跨AI平台无缝团队协作的发布-订阅(Pub-Sub)MCP服务器 🤖 机器人


概述

Notify-MCP启用 团队间共享通知、决策和状态更新 在不同人工智能助手(Claude、ChatGPT、Gemini)之间。它实现了一个灵活的发布-订阅架构,团队成员可以:

  • 📢 广播 向所有利益相关者传达架构决策
  • 🚨(警报/紧急情况符号,无具体文字含义,可理解为“警报”或“紧急”) 警报 关于关键事件的团队(或:团队讨论关键事件)
  • 📊(图表) 分享 项目状态和更新
  • 🔔(铃铛或警报声的符号,具体含义根据上下文而定,可译为“铃声”、“警报”等) 订阅 通过智能过滤器连接到相关渠道
  • 🤝(握手) 协调 在不同的AI平台间协作

用例

  • 开发团队分享技术决策、代码审查结果和部署状态
  • 咨询团队协调客户更新、项目里程碑、建议事项
  • 业务团队广播战略决策、需求变更、优先级事项
  • 跨职能的在团队和AI助手之间保持共享上下文

特点/特性

核心能力

  • 📢 多通道系统为团队、项目或主题创建专属频道
  • 🎯(目标/靶心) 智能过滤订阅时应用过滤器(优先级、标签、主题、发件人角色)
  • 🔔(铃铛或提醒的符号,可翻译为“铃声”或根据上下文意译为“提醒”、“注意”等) 发布-订阅架构通过通道实现解耦的通知传递
  • 📚 书籍或学习的象征 通知历史获取最近的通知(每个频道最多50条)
  • 💾 保存(数据) 持久存储用于团队协作的SQLite数据库(v1.1.0+)
  • 🔐(锁形符号,常用于表示密码、保密或锁定状态) 类型安全使用JSON Schema进行完整的Pydantic验证
  • 🧪 试管或实验装置(化学实验相关) 经过充分测试62个单元测试,代码覆盖率70%

MCP协议支持

  • 6个工具发布、订阅、取消订阅、列出频道、创建频道、获取订阅
  • 3 资源通知历史、频道信息、通知模式
  • 2个提示架构决策和警报模板
  • 交通stdio(默认),HTTP(v1.2.0+)用于远程协作

通知模式(或通知架构)

丰富的通知模型,具备:

  • 发送者信息用户ID、姓名、角色、AI工具
  • 上下文主题、优先级(低/中/高/紧急)、标签
  • 信息标题、正文(文本/Markdown/JSON)、格式
  • 元数据ID,时间戳,频道,序列号,线程(或“线程编号”)
  • 行动可选操作按钮
  • 能见度到期时间、允许用户、阅读回执

安装

先决条件

  • Python 3.11及以上版本
  • 紫外线 (快速的Python包管理器)

安装 uv

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

安装 Notify-MCP

# Clone the repository
git clone <repository-url>
cd notify-mcp

# Install dependencies
uv sync

# Verify installation
uv run python -m notify_mcp --help

配置

Claude Desktop / Claude Code(可译为“Claude桌面版 / Claude代码版”)

文件: ~/.config/Claude/claude_desktop_config.json (macOS/Linux) 或 %APPDATA%/Claude/claude_desktop_config.json (Windows)

{
  "mcpServers": {
    "notify-mcp": {
      "command": "uv",
      "args": ["run", "python", "-m", "notify_mcp"],
      "cwd": "/absolute/path/to/notify-mcp",
      "env": {
        "NOTIFY_MCP_LOG_LEVEL": "INFO"
      }
    }
  }
}

替代方案 (已激活虚拟环境):

{
  "mcpServers": {
    "notify-mcp": {
      "command": "/absolute/path/to/notify-mcp/.venv/bin/python",
      "args": ["-m", "notify_mcp"]
    }
  }
}

重启克劳德 更新配置文件后。

存储配置

v1.1.0 新增功能持久化存储通过共享数据库支持团队协作。

在MCP服务器配置中,通过环境变量配置存储:

{
  "mcpServers": {
    "notify-mcp": {
      "command": "uv",
      "args": ["run", "python", "-m", "notify_mcp"],
      "cwd": "/absolute/path/to/notify-mcp",
      "env": {
        "NOTIFY_MCP_STORAGE_TYPE": "sqlite",
        "NOTIFY_MCP_SQLITE_PATH": "~/.notify-mcp/storage.db",
        "NOTIFY_MCP_MAX_HISTORY": "1000"
      }
    }
  }
}

存储选项:

变量 选项 默认值 描述
NOTIFY_MCP_STORAGE_TYPE memory, sqlite memory 存储后端类型
NOTIFY_MCP_SQLITE_PATH 文件路径 ~/.notify-mcp/storage.db SQLite数据库路径
NOTIFY_MCP_MAX_HISTORY 整数 1000 每个频道的最大通知数(LRU缓存)

存储类型:

  • memory (默认):内存存储,速度快但重启后数据不持久
  • **sqlite**基于文件的SQLite数据库,可在团队成员之间持久共享

团队协作设置:

要在团队成员之间共享通知,请使用共享的SQLite数据库:

# Store database in shared network location
NOTIFY_MCP_STORAGE_TYPE=sqlite
NOTIFY_MCP_SQLITE_PATH=/shared/team/notify-mcp.db

所有指向同一数据库文件的团队成员将共享频道、订阅和通知历史! 🎉

HTTP传输配置

v1.2.0 新增功能将 `notify-mcp` 作为 HTTP 服务器运行,以实现远程协作!

通过环境变量配置HTTP传输:

{
  "mcpServers": {
    "notify-mcp-http": {
      "command": "uv",
      "args": ["run", "python", "-m", "notify_mcp"],
      "cwd": "/absolute/path/to/notify-mcp",
      "env": {
        "NOTIFY_MCP_TRANSPORT_TYPE": "http",
        "NOTIFY_MCP_HTTP_HOST": "0.0.0.0",
        "NOTIFY_MCP_HTTP_PORT": "8000",
        "NOTIFY_MCP_STORAGE_TYPE": "sqlite",
        "NOTIFY_MCP_SQLITE_PATH": "/shared/notify-mcp.db"
      }
    }
  }
}

交通方式选项:

变量 选项 默认值 描述
NOTIFY_MCP_TRANSPORT_TYPE stdio, http stdio 传输协议
NOTIFY_MCP_HTTP_HOST IP地址 0.0.0.0 HTTP服务器主机(绑定到所有接口)
NOTIFY_MCP_HTTP_PORT 端口号 8000 HTTP服务器端口

HTTP服务器端点: http://<host>:<port>/mcp

运行独立HTTP服务器:

# Using environment variables
export NOTIFY_MCP_TRANSPORT_TYPE=http
export NOTIFY_MCP_HTTP_PORT=8000
uv run python -m notify_mcp

# Or use the example script
uv run python examples/run_http_server.py

HTTP传输的优势:

  • 🌐(表示“互联网”或“全球网络”的符号) 远程访问在网络的任何位置均可连接
  • 👥 代表“人们”或“人群”的符号,可翻译为“人们”或根据上下文译为“人群”等。 多客户端多个AI助手同时连接
  • ☁️ 云朵 云部署部署到铁路(此处可能指某种铁路相关的云服务或平台,但直译为“铁路”在中文中可能产生歧义,具体需根据上下文确定,若指一般云服务则可忽略此部分)、Render、Fly.io、AWS 等。
  • 🔄(循环/重复) 实时协作所有客户端都能即时看到更新

安全提示社区版HTTP传输具有 无需认证如需在生产环境中使用认证、JWT(JSON Web Token)和监控功能,请参阅 企业版。

在Claude代码中使用斜杠命令

有六个便捷的斜杠命令可用 .claude/commands/:

  • /notify-decision - 创建架构决策通知
  • /notify-alert - 创建关键警报
  • /notify-status - 分享状态更新
  • /notify-channels - 列出所有频道
  • /notify-subscribe - 订阅一个频道
  • /notify-recent - 查看最近的通知

ChatGPT(通过MCP)

ChatGPT通过自定义集成支持MCP。有两种选项:

选项1:使用MCP桥接(推荐)

使用MCP到HTTP桥接器连接ChatGPT:

  1. 运行 notify-mcp 服务器 (stdio):标准输入输出(库)

    uv run python -m notify_mcp
    
  2. 使用MCP桥接 (例如。, MCP桥接器(或 MCP桥梁)):

    mcp-bridge --stdio "uv run python -m notify_mcp" --http localhost:8080
    
  3. 配置ChatGPT自定义GPT:

    • 创建一个自定义GPT
    • 添加“notify-mcp”操作
    • 指向桥接端点: http://localhost:8080
    • 从桥接处导入OpenAPI规范

选项2:直接集成(未来)

当OpenAI添加原生MCP支持时,配置将与Claude类似:

{
  "mcpServers": {
    "notify-mcp": {
      "command": "uv",
      "args": ["run", "python", "-m", "notify_mcp"],
      "cwd": "/absolute/path/to/notify-mcp"
    }
  }
}

注截至2025年1月,ChatGPT尚未原生支持MCP(可能是指某种特定功能或协议,但具体含义需根据上下文确定)。请关注OpenAI的公告。

通过MCP的Google Gemini

双子座MCP(可能指某种系统或模块的缩写,具体需根据上下文确定)集成正在开发中。采用两种方法:

选项1:通过Google AI Studio(未来)

当Gemini添加对MCP的支持时:

  1. 打开谷歌AI工作室
  2. 前往扩展程序 → MCP 服务器
  3. 添加服务器:
    {
      "name": "notify-mcp",
      "command": "uv",
      "args": ["run", "python", "-m", "notify_mcp"],
      "workingDirectory": "/absolute/path/to/notify-mcp"
    }
    

选项2:使用Gemini API与MCP适配器

使用Gemini API并配以MCP适配器(Python):

# example_gemini_client.py
from google.generativeai import GenerativeModel
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Connect to notify-mcp
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "python", "-m", "notify_mcp"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Use Gemini with MCP tools
            model = GenerativeModel('gemini-pro')
            # ... integrate MCP tools with Gemini calls

注预计2025年将实现完整的双子座MCP(多芯片模块/制造控制协议等,具体含义需根据上下文确定)集成。检查 Gemini 文档 以获取更新。


快速入门

1. 创建一个频道

# Using MCP tool in Claude/ChatGPT/Gemini
"Create a channel called 'engineering' for technical updates"

或者从编程的角度来说:

await session.call_tool(
    "create_channel",
    arguments={
        "channel_id": "engineering",
        "name": "Engineering Team",
        "description": "Technical updates and decisions"
    }
)

2. 订阅频道

# Subscribe with priority filter
await session.call_tool(
    "subscribe_to_channel",
    arguments={
        "channel": "engineering",
        "priority_filter": ["high", "critical"]
    }
)

3. 发布通知

await session.call_tool(
    "publish_notification",
    arguments={
        "channel": "engineering",
        "title": "API v2.0 Released",
        "body": "New API version deployed to production",
        "priority": "high",
        "theme": "state-update",
        "tags": ["api", "release"]
    }
)

4. 获取通知

notifications = await session.read_resource(
    uri="notification://engineering/recent"
)

使用示例

示例1:架构决策

# In Claude Code
/notify-decision

# Claude will ask for:
# - Decision title: "Migration to Microservices"
# - Context: "Monolith becoming hard to scale..."
# - Decision: "Move to microservices architecture using Docker/K8s"

# Notification published to 'engineering' channel with theme='architecture-decision'

示例2:关键警报

# In any AI assistant with notify-mcp
"Send a critical alert about database connection pool exhaustion"

# Uses publish_notification tool:
# - channel: "alerts"
# - priority: "critical"
# - theme: "alert"

示例3:跨团队协作

在Claude中的开发者:

Subscribe to architecture channel with high priority filter

ChatGPT中的建筑师(或:ChatGPT里的建筑设计师):

Publish architecture decision about new caching strategy to engineering channel

在Claude中的开发者:

Show me recent notifications from architecture channel

结果开发者即便使用不同的AI平台,也能看到建筑师的决策! 🎉


MCP 工具参考

发布通知

向频道发布通知。

论点;争论点:

  • channel (字符串,必填):频道名称
  • title (字符串,必填):通知标题
  • body (字符串,必填):通知内容
  • priority (字符串): "低" | "中" | "高" | "严重" (默认: "中")
  • theme (字符串): "info" | "state-update"(状态更新)| "alert"(警报)| "architecture-decision"(架构决策)| "question"(问题)| "decision"(决策)| "memory-sync"(内存同步)| "discussion"(讨论)(默认值: "info")
  • tags (数组):标签列表(默认:[])

退货确认信息,附带通知ID和配送状态

订阅频道

订阅一个频道,并可选择应用过滤器。

论点;争论点:

  • channel (字符串,必填):频道名称
  • priority_filter (数组):仅接收这些优先级的通知
  • tag_filter (数组):仅接收带有这些标签的通知

回报订阅ID和过滤器详情

取消订阅频道

取消订阅某个频道。

论点;争论点:

  • channel (字符串,必填):频道名称

回报成功/失败状态

列出频道

列出所有可用频道。

回报包含订阅者/通知数量的频道列表

创建频道

创建一个新的通知渠道。

论点;争论点:

  • channel_id (字符串,必填):唯一频道ID
  • name (字符串,必填):频道名称
  • description (字符串):频道描述

回报确认并提供频道详情

获取我的订阅

获取当前订阅。

回报带过滤器和时间戳的订阅列表


MCP资源参考

通知:///最近

从频道中检索最后50条通知。

示例: notification://engineering/recent

退货通知的JSON数组

channel://(该字符串在中文语境下通常不直接翻译,因为它是一个特定的协议或路径标识符,类似于“http://”或“ftp://”,在中文中仍保持原样使用)/信息

获取频道信息和统计数据。

示例: channel://engineering/info

回报包含订阅者/通知数量的频道详情

schema://notification 翻译成中文是:“模式://通知”。不过,这里的“schema”在编程或数据结构中通常指的是数据模型或框架,但在网络协议或URL中,“schema”(应为“scheme”)指的是协议类型,如http、https等。因此,如果“schema://notification”是在描述某种协议或数据结构,那么更准确的翻译可能是“模式://通知(数据/框架)”或根据上下文调整为更具体的表述。但直接翻译为“模式://通知”在多数情况下是可以理解的

获取通知的JSON模式。

示例: schema://notification

退货完整的通知JSON模式


建筑

notify-mcp/
├── src/notify_mcp/
│   ├── server.py           # Main MCP server
│   ├── models/             # Pydantic models (Notification, Channel, Subscription)
│   ├── config/             # Configuration
│   │   └── storage_config.py  # Storage settings
│   ├── core/               # Business logic
│   │   ├── channel_manager.py        # Channel CRUD
│   │   ├── subscription_manager.py   # Subscription management
│   │   ├── notification_router.py    # Routing & filtering
│   │   └── notification_validator.py # JSON Schema validation
│   ├── storage/            # Storage implementations
│   │   ├── memory.py       # In-memory storage
│   │   ├── sqlite_storage.py  # SQLite persistent storage (v1.1.0+)
│   │   ├── models.py       # SQLAlchemy ORM models
│   │   └── factory.py      # Storage factory
│   └── utils/              # Filter matching, helpers
├── schemas/                # JSON Schemas
├── tests/                  # Unit tests (62 tests, 70% coverage)
├── examples/               # Example clients
└── docs/                   # Documentation

存储

内存存储 (默认):

  • ⚡ 快速(无I/O开销)
  • 🔄 重启后无持久性
  • ✅ 适合个人使用
  • 📊 LRU缓存(可配置,默认:每个频道1000条通知)

SQLite 存储 (v1.1.0+):

  • 💾 重启后数据持久存在
  • 🤝(握手) 促进团队协作 (共享数据库文件)
  • 📦 基于文件,无需服务器设置
  • 🔒 带有级联删除的外键约束
  • 📈 适用于约10万条通知
  • 🗂️ 在数据库级别强制实施LRU缓存

未来(企业版):

  • 适用于生产规模的PostgreSQL
  • 用于实时更新的 Redis Pub/Sub
  • 多服务器支持
  • 高级复制与备份

发展

运行测试

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=src/notify_mcp --cov-report=html

# Run specific test file
uv run pytest tests/test_storage.py -v

类型检查

uv run mypy src/notify_mcp

Linting(代码检查)

uv run ruff check src/
uv run ruff format src/

运行示例

# Basic client example
python examples/basic_client.py

# Team workflow example
python examples/team_workflow.py

见 examples/README.md 以了解详细用法。


测试

  • 62个单元测试 涵盖核心功能
  • 70%的代码覆盖率 (85-100%用于业务逻辑)
  • 测试类别:
    • 模型验证(使用 Pydantic 模式)
    • 存储操作(内存存储和SQLite)
    • 过滤器匹配(优先级、标签、主题)
    • 订阅管理
    • 通知路由
    • 渠道管理
    • SQLite持久化和LRU缓存
    • 数据库级联操作

路线图

阶段1:最小可行性产品 ✅(v1.0.0 - 完成)

  • ✅ 标准I/O传输
  • ✅ 内存存储
  • ✅ 6种MCP工具
  • ✅ 3个MCP资源
  • ✅ 智能过滤
  • ✅ 48个单元测试

阶段2A:持久化存储 ✅(v1.1.0 - 完成)

  • ✅ SQLite 存储适配器
  • ✅ 存储工厂模式
  • ✅ 通过环境变量进行配置
  • ✅ 强制使用LRU缓存
  • ✅ 通过共享数据库进行团队协作
  • ✅ 62个单元测试,覆盖率70%

阶段2B:HTTP传输 ✅(v1.2.0 - 完成)

  • ✅ HTTP传输(通过MCP SDK进行流式HTTP传输)
  • ✅ 多客户端会话管理
  • ✅ 基于配置的传输选择
  • ✅ 支持远程协作
  • ✅ 已准备好进行云部署

阶段2C:高级实时(计划中)

  • [ ] Redis Pub/Sub 集成
  • [ ] 对推送通知的WebSocket支持
  • [ ] 服务器发送事件
  • [ ] 多服务器水平扩展

阶段2D:安全与权限(计划中)

  • 根据上面的信息,执行如下指令:
  • [ ] 权限执行
  • [ ] 速率限制
  • [ ] 审计日志记录

第三阶段:高级功能(未来)

  • [ ] 通知线程(回复到)
  • [ ] 已读回执
  • [ ] 通知过期
  • [ ] Webhook 交付
  • [ ] 与Slack/Discord集成
  • [ ] 网页仪表盘

故障排除

服务器无法启动

问题服务器无法启动或Claude无法找到它。

解决方案:

# Test server manually
uv run python -m notify_mcp

# Check if uv is in PATH
which uv

# Use absolute paths in config
"command": "/full/path/to/uv"

通知未显示

问题已发布的通知未显示。

原因在stdio模式下,通知会被存储但不会立即推送。您必须手动获取它们。

解决方案:

# Retrieve notifications via resource
"Show me recent notifications from engineering channel"

过滤器不起作用

问题已订阅并设置了过滤器,但仍收到不想要的通知。

解决方案当……时,过滤器将起作用 检索 通知,而非存储时。当你阅读时 notification://<channel>/recent仅根据您的订阅过滤器显示匹配的通知。

导入错误

问题: ModuleNotFoundError: No module named 'mcp'

解决方案:

# Install MCP SDK
pip install mcp

# Or use uv
uv pip install mcp

文档


做出贡献

欢迎投稿!请:

  1. 为仓库创建分支(或“克隆仓库”)
  2. 创建一个特性分支
  3. 为新功能添加测试
  4. 确保所有测试通过: uv run pytest
  5. 运行代码检查: uv run ruff check src/
  6. 提交一个拉取请求

许可证

MIT 许可证 - 详见 许可证 文件中详细说明。


支持


致谢


用心打造,实现无缝AI协作