notifications-mcp
Streamable HTTP + SSE MCP 服务器
生产就绪的模型上下文协议 (MCP) 服务器,支持 Streamable HTTP + SSE、Bearer 认证、严格会话身份绑定、提示注入防护、收件人白名单和速率限制。
👥 分工与职责
贡献者 | 领域与交付物 |
@Vinod | 服务器脚手架、Streamable HTTP( |
@Vishal | 工具处理程序( |
Related MCP server: Gmail MCP Server
🌟 核心功能与标准
官方 MCP 规范标准:基于 Streamable HTTP(
POST /mcp和GET /mcpSSE)的 JSON-RPC 2.0 传输。会话安全与断线重连恢复:
Mcp-Session-Id:UUID 会话,严格绑定到已认证的调用者身份。Last-Event-ID:SSE 事件环形缓冲区,用于重连时的事件重放。
安全防护:
认证强制执行:每个
/mcp请求必须携带有效的 Bearer 令牌或 API 密钥。会话劫持防御:会话 ID 不能在不同身份之间共享。
提示注入防御:使用正则和启发式分析器扫描电子邮件字段,并记录安全审计日志。
收件人白名单与速率限制:防止未经授权的电子邮件外泄和泛滥。
干净的错误消毒:通用、安全的 JSON-RPC 错误(不泄露堆栈跟踪)。
模拟工具:
get_current_datetime(timezone?):系统日期/时间,支持 IANA 时区验证。send_email(to, subject, body):服务器 SMTP,带安全模拟演示模式。
📁 仓库结构
Custom_MCP_Server/
├── config.yaml # Server, auth tokens, allowlists, rate limits, and SMTP config
├── pyproject.toml # Package definition & dependencies
├── requirements.txt # Dependencies
├── README.md # Documentation & demo checklist
├── scripts/
│ ├── run_tunnel.py # Hosting / tunnel launcher (Cloudflared, Ngrok, Localtunnel, Local)
│ └── test_demo_cases.py # Automated verification suite for the 6 Live Demo test cases
├── src/
│ ├── __init__.py
│ ├── config.py # Typed configuration models
│ ├── server.py # Starlette Streamable HTTP + SSE application
│ ├── registry.py # MCP tools registration & dispatcher
│ ├── auth/
│ │ ├── __init__.py
│ │ ├── middleware.py # Auth, Origin, and session validation middleware
│ │ └── session_manager.py # UUID session manager with Last-Event-ID buffer
│ ├── security/
│ │ ├── __init__.py
│ │ ├── injection_detector.py # Prompt injection detector & audit logger
│ │ ├── allowlist.py # Email recipient & domain allowlist validator
│ │ └── rate_limiter.py # Sliding-window rate limiter per caller
│ ├── common/
│ │ ├── __init__.py
│ │ ├── errors.py # JSON-RPC 2.0 error codes and builders
│ │ └── logging.py # Structured JSON logger with security tags
│ └── tools/
│ ├── __init__.py
│ ├── datetime_tool.py # get_current_datetime tool handler
│ └── email_tool.py # send_email tool handler with SMTP/demo mode
└── tests/
├── __init__.py
├── conftest.py
├── test_datetime_tool.py # Datetime unit tests
├── test_email_tool.py # Email & security guardrail tests
├── test_security_auth.py # Auth & session isolation tests
└── test_streamable_http.py # Streamable HTTP / SSE transport tests🚀 快速入门指南
1. 前提条件
Python 3.10+
Node.js(可选,用于 MCP Inspector)
2. 安装依赖
pip install -r requirements.txt3. 启动服务器
python -m src.server您将看到:
{"timestamp": "...", "level": "INFO", "message": "Starting Streamable MCP Server | host=0.0.0.0 | port=8100"}
{"timestamp": "...", "level": "INFO", "message": "MCP POST/GET endpoint → http://localhost:8100/mcp"}
{"timestamp": "...", "level": "INFO", "message": "Health check probe → http://localhost:8100/health"}⚙️ 配置 (config.yaml)
server:
host: "0.0.0.0"
port: 8100
allowed_origins:
- "http://localhost:8100"
- "*"
auth:
enabled: true
tokens:
"agent-token-alpha": "agent_alpha"
"agent-token-beta": "agent_beta"
"vishal-test-token": "vishal_engineer"
"vinod-test-token": "vinod_engineer"
security:
email:
allowlist:
- "ops@company.com"
- "manager@company.com"
- "vishal@company.com"
- "vinod@company.com"
- "*@trusteddomain.com"
rate_limit:
max_calls: 10
window_seconds: 60
prompt_injection_guard:
enabled: true
smtp:
mode: "simulation" # "simulation" (safe for testing/demos) or "smtp" (live server)
host: "smtp.example.com"
port: 587
sender_address: "notifications@company.com"🛠️ 工具模式
工具 1: get_current_datetime
描述: 返回当前主机系统日期和时间,支持可选的 IANA 时区转换。
参数:
{ "type": "object", "properties": { "timezone": { "type": "string", "description": "Optional IANA timezone name (e.g. 'UTC', 'America/New_York', 'Asia/Kolkata')." } }, "required": [] }
工具 2: send_email
描述: 通过服务器 SMTP 发送电子邮件通知,带有严格的安全防护。
参数:
{ "type": "object", "properties": { "to": { "type": "string", "description": "Recipient email address. Must be in the authorized allowlist." }, "subject": { "type": "string", "description": "Email subject line." }, "body": { "type": "string", "description": "Email body content." } }, "required": ["to", "subject", "body"] }
🔌 连接 MCP Inspector 与 cURL
通过 MCP Inspector 连接:
npx @modelcontextprotocol/inspector连接到 URL:http://localhost:8100/mcp,自定义 Header:Authorization: Bearer vishal-test-token。
示例 cURL 调用:
1. 调用 get_current_datetime:
curl -X POST http://localhost:8100/mcp \
-H "Content-Type: application/json" \
-H "Authorization: Bearer vishal-test-token" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "get_current_datetime",
"arguments": {"timezone": "Asia/Kolkata"}
}
}'2. 调用 send_email:
curl -X POST http://localhost:8100/mcp \
-H "Content-Type: application/json" \
-H "Authorization: Bearer vishal-test-token" \
-d '{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "send_email",
"arguments": {
"to": "ops@company.com",
"subject": "System Status",
"body": "All services operating normally."
}
}
}'🌐 托管与隧道设置
将 MCP 服务器暴露到公网 URL / 云端演示:
# Option 1: Direct Local / VM binding
python scripts/run_tunnel.py --mode local --port 8100
# Option 2: Cloudflare Tunnel (Free, no port-forwarding needed)
python scripts/run_tunnel.py --mode cloudflared --port 8100
# Option 3: Ngrok
python scripts/run_tunnel.py --mode ngrok --port 8100
# Option 4: Localtunnel
python scripts/run_tunnel.py --mode localtunnel --port 8100🧪 现场演示清单(6 个测试用例)
一键运行自动化测试运行器,验证全部 6 个演示用例:
python scripts/test_demo_cases.py或者使用 pytest 运行:
pytest -v已验证的测试用例:
用例 1:授权调用每个工具成功(
get_current_datetime和send_email均通过)。用例 2:未授权调用者收到通用错误(HTTP 401,错误代码
-32001)。用例 3:格式错误的输入返回干净的验证错误(无效时区或缺少模式字段)。
用例 4:电子邮件中的提示注入尝试被阻止并记录(例如系统提示覆盖)。
用例 5:发送到非白名单电子邮件被拒绝(阻止外部/未授权地址)。
用例 6:跨不同身份重用会话 ID 被拒绝(HTTP 403,错误代码
-32005)。
This server cannot be deployed
Maintenance
Related MCP Connectors
MCP server connecting AI agents to 100+ apps (Gmail, Slack, Notion, GitHub) via one-click OAuth.
Email infrastructure for AI agents — send, receive, search, and reply to email over MCP.
Unified MCP Server is a remote MCP connector for AI agents and vertical AI products that provides access to 22,000+ authorized SaaS tools across 400+ integrations and 24 categories directly inside LLMs (Claude, GPT, Gemini, Cohere). Tools operate only on explicitly authorized customer connections, enabling agents to safely read and write against live third-party systems.
The Mobile Text Alerts SMS MCP server enables your AI to send SMS messages & manage contacts
Related MCP Servers
- AlicenseNot gradedqualityDmaintenanceAn MCP server that provides AI applications with a toolset for an Artificial Virtual Assistant (AVA) integrated with the Gmail API. It enables users to automate email management and personal tasks through secure Google OAuth authentication.1Apache 2.0
- AlicenseNot gradedqualityDmaintenanceA Node.js server that connects MCP to Gmail, enabling LLMs to send, read, and manage emails securely using the Gmail API.184 npm2MIT
- FlicenseNot gradedqualityDmaintenanceEnables AI agents to interact with Gmail through a standardized MCP server interface, allowing for natural language email management and automation.-
- FlicenseNot gradedqualityDmaintenanceProduction-ready MCP server for Gmail, enabling AI agents to search, read, send, draft, and manage emails, labels, and attachments via the Google Gmail API.-