Промышленный MCP-сервер (Model Context Protocol) с Streamable HTTP + SSE, Bearer-аутентификацией, строгой привязкой сессии к идентификатору, защитой от инъекций в промпты, белым списком получателей и ограничением скорости.
Участник | Область и результаты |
@Vinod | Создание каркаса сервера, транспорт Streamable HTTP ( |
@Vishal | Обработчики инструментов ( |
Related MCP server: Gmail MCP Server
Официальный стандарт MCP-спецификации: транспорт JSON-RPC 2.0 поверх Streamable HTTP (POST /mcp и GET /mcp SSE).
Безопасность сессий и возобновляемость:
Mcp-Session-Id: UUID-сессии, строго привязанные к аутентифицированным вызывающим сторонам.
Last-Event-ID: кольцевой буфер SSE-событий для повторного воспроизведения событий при переподключении.
Защитные механизмы:
Принудительная аутентификация: каждый запрос к /mcp требует корректный Bearer-токен или API-ключ.
Защита от угона сессии: идентификаторы сессий не могут использоваться разными сторонами.
Защита от инъекций в промпты: анализ email-полей с помощью регулярных выражений и эвристик с записью в аудит-лог безопасности.
Белый список получателей и ограничение скорости: предотвращает несанкционированную утечку email-адресов и флуд.
Чистая санитизация ошибок: обобщённые, безопасные ошибки 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 testsPython 3.10+
Node.js (опционально, для MCP Inspector)
pip install -r requirements.txtpython -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"get_current_datetimeОписание: Возвращает текущие системные дату и время хоста с опциональным преобразованием в часовой пояс IANA.
Параметры:
{
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Optional IANA timezone name (e.g. 'UTC', 'America/New_York', 'Asia/Kolkata')."
}
},
"required": []
}send_emailОписание: Отправляет 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"]
}npx @modelcontextprotocol/inspectorПодключитесь к URL: http://localhost:8100/mcp с пользовательским заголовком: Authorization: Bearer vishal-test-token.
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 демо-сценариев:
python scripts/test_demo_cases.pypytest -vСценарий 1: Авторизованный вызов каждого инструмента успешен (оба get_current_datetime и send_email).
Сценарий 2: Неавторизованный вызывающий получает обобщённую ошибку (HTTP 401, код ошибки -32001).
Сценарий 3: Некорректный ввод возвращает чистую ошибку валидации (неверный часовой пояс или отсутствующие поля схемы).
Сценарий 4: Попытки инъекции промптов в email блокируются и логируются (например, переопределение системного промпта).
Сценарий 5: Отправка на email не из белого списка отклоняется (блокируются внешние/неавторизованные адреса).
Сценарий 6: Повторное использование идентификатора сессии разными сторонами запрещено (HTTP 403, код ошибки -32005).
This server cannot be deployed
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
