Servidor de Protocolo de Contexto de Modelo (MCP) listo para producción con HTTP Streamable + SSE, Autenticación Bearer, Vinculación Estricta de Identidad de Sesión, Guardas contra Inyección de Prompts, Lista Blanca de Destinatarios y Límite de Tasa.
Colaborador | Área y Entregables |
@Vinod | Andamiaje del servidor, transporte HTTP Streamable ( |
@Vishal | Manejadores de herramientas ( |
Related MCP server: Gmail MCP Server
Estándar Oficial de Especificación MCP: Transporte JSON-RPC 2.0 sobre HTTP Streamable (POST /mcp y GET /mcp SSE).
Seguridad de Sesión y Reanudabilidad:
Mcp-Session-Id: Sesiones UUID estrictamente vinculadas a identidades de llamante autenticadas.
Last-Event-ID: Búfer circular de eventos SSE para reproducción de eventos al reconectar.
Guardas de Seguridad:
Cumplimiento de Autenticación: Cada solicitud /mcp requiere un token Bearer o clave API válida.
Defensa contra Secuestro de Sesión: Los IDs de sesión no pueden compartirse entre diferentes identidades.
Defensa contra Inyección de Prompts: Analizador de expresiones regulares y heurístico que escanea campos de correo electrónico con registro de auditoría de seguridad.
Lista Blanca de Destinatarios y Límites de Tasa: Previene la exfiltración no autorizada de correos electrónicos y la saturación.
Limpieza de Errores: Errores JSON-RPC genéricos y seguros (sin fugas de trazas de pila).
Herramientas Simuladas:
get_current_datetime(timezone?): Fecha/hora del sistema con validación de zona horaria IANA.
send_email(to, subject, body): SMTP del servidor con modo de demostración de simulación segura.
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 (opcional, para MCP Inspector)
pip install -r requirements.txtpython -m src.serverVerás:
{"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_datetimeDescripción: Devuelve la fecha y hora actual del sistema anfitrión con conversión opcional de zona horaria IANA.
Parámetros:
{
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Optional IANA timezone name (e.g. 'UTC', 'America/New_York', 'Asia/Kolkata')."
}
},
"required": []
}send_emailDescripción: Envía una notificación por correo electrónico a través del SMTP del servidor con estrictas guardas de seguridad.
Parámetros:
{
"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/inspectorConectar a la URL: http://localhost:8100/mcp con el encabezado personalizado: Authorization: Bearer vishal-test-token.
1. Llamar a 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. Llamar a 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."
}
}
}'Para exponer el servidor MCP a una URL pública / demostración en la nube:
# 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 8100Ejecuta el ejecutor de pruebas automatizado para verificar los 6 casos de demostración en un solo comando:
python scripts/test_demo_cases.pypytest -vCaso 1: Llamada autorizada a cada herramienta exitosa (tanto get_current_datetime como send_email).
Caso 2: Llamante no autorizado recibe un error genérico (HTTP 401, código de error -32001).
Caso 3: Entrada malformada devuelve un error de validación limpio (zona horaria inválida o campos de esquema faltantes).
Caso 4: Intentos de inyección de prompts en el correo electrónico son bloqueados y registrados (por ejemplo, anulaciones de prompt del sistema).
Caso 5: Envío a un correo electrónico no incluido en la lista blanca es rechazado (bloquea direcciones externas/no autorizadas).
Caso 6: Reutilización de ID de sesión entre diferentes identidades es denegada (HTTP 403, código de error -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
