使用说明
通知-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:
-
运行 notify-mcp 服务器 (stdio):标准输入输出(库)
uv run python -m notify_mcp -
使用MCP桥接 (例如。, MCP桥接器(或 MCP桥梁)):
mcp-bridge --stdio "uv run python -m notify_mcp" --http localhost:8080 -
配置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的支持时:
- 打开谷歌AI工作室
- 前往扩展程序 → MCP 服务器
- 添加服务器:
{ "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(字符串,必填):唯一频道IDname(字符串,必填):频道名称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
文档
- **API 文档**完整的MCP API参考
- **通知模式(或通知架构)**模式规范
- **建筑**系统设计
- **使用指南**详细使用场景
- **示例**代码示例
做出贡献
欢迎投稿!请:
- 为仓库创建分支(或“克隆仓库”)
- 创建一个特性分支
- 为新功能添加测试
- 确保所有测试通过:
uv run pytest - 运行代码检查:
uv run ruff check src/ - 提交一个拉取请求
许可证
MIT 许可证 - 详见 许可证 文件中详细说明。
支持
- 问题: GitHub 问题(或“GitHub 问题跟踪”)
- 文档: 文档/
- 示例: 示例/例子/
致谢
- 构建于 MCP(模型上下文协议) 由Anthropic公司(或团队)开发/推出
- 用途 Pydantic(注:这是一个用于数据验证和设置管理的Python库,常用于构建API接口等场景) 用于数据验证
- 通过(使用)包管理器进行管理 紫外线
用心打造,实现无缝AI协作