Skip to main content
Glama
Manojbonthu

notifications-mcp

by Manojbonthu

Streamable HTTP + SSE MCP 服务器

生产就绪的模型上下文协议 (MCP) 服务器,支持 Streamable HTTP + SSE、Bearer 认证、严格会话身份绑定、提示注入防护、收件人白名单和速率限制。


👥 分工与职责

贡献者

领域与交付物

@Vinod

服务器脚手架、Streamable HTTP(POST + GET SSE)传输层、Mcp-Session-Id 会话管理、Last-Event-ID 断线重连恢复、托管与隧道设置(scripts/run_tunnel.py)、Bearer 令牌认证、来源检查、JSON-RPC 错误处理层。

@Vishal

工具处理程序(get_current_datetime、send_email)、输入验证与 Pydantic 模式、提示注入检测与审计日志、收件人白名单与滑动窗口速率限制、6 个现场演示验证测试用例(scripts/test_demo_cases.py)。


Related MCP server: Gmail MCP Server

🌟 核心功能与标准

  1. 官方 MCP 规范标准:基于 Streamable HTTP(POST /mcp 和 GET /mcp SSE)的 JSON-RPC 2.0 传输。

  2. 会话安全与断线重连恢复:

    • Mcp-Session-Id:UUID 会话,严格绑定到已认证的调用者身份。

    • Last-Event-ID:SSE 事件环形缓冲区,用于重连时的事件重放。

  3. 安全防护:

    • 认证强制执行:每个 /mcp 请求必须携带有效的 Bearer 令牌或 API 密钥。

    • 会话劫持防御:会话 ID 不能在不同身份之间共享。

    • 提示注入防御:使用正则和启发式分析器扫描电子邮件字段,并记录安全审计日志。

    • 收件人白名单与速率限制:防止未经授权的电子邮件外泄和泛滥。

    • 干净的错误消毒:通用、安全的 JSON-RPC 错误(不泄露堆栈跟踪)。

  4. 模拟工具:

    • 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.txt

3. 启动服务器

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)。

Related MCP Connectors

Related MCP Servers

  • A
    license
    Not graded
    quality
    D
    maintenance
    An 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.
    1
    Apache 2.0
  • A
    license
    Not graded
    quality
    D
    maintenance
    A Node.js server that connects MCP to Gmail, enabling LLMs to send, read, and manage emails securely using the Gmail API.
    184 npm
    2
    MIT
  • F
    license
    Not graded
    quality
    D
    maintenance
    Production-ready MCP server for Gmail, enabling AI agents to search, read, send, draft, and manage emails, labels, and attachments via the Google Gmail API.
    -