Produktionsreifer Model Context Protocol (MCP) Server mit Streamable HTTP + SSE, Bearer-Authentifizierung, strenger Sitzungsidentitätsbindung, Prompt-Injection-Schutzmechanismen, Empfänger-Whitelisting und Ratenbegrenzung.
Mitwirkender | Bereich & Leistungen |
@Vinod | Server-Grundgerüst, Streamable HTTP ( |
@Vishal | Tool-Handler ( |
Related MCP server: Gmail MCP Server
Offizieller MCP-Spezifikationsstandard: JSON-RPC 2.0 Transport über Streamable HTTP (POST /mcp und GET /mcp SSE).
Sitzungssicherheit & Wiederaufnahmefähigkeit:
Mcp-Session-Id: UUID-Sitzungen, die streng an authentifizierte Aufruferidentitäten gebunden sind.
Last-Event-ID: SSE-Ereignisringpuffer für Ereigniswiederholung bei erneuter Verbindung.
Sicherheitsschutzmechanismen:
Authentifizierungsdurchsetzung: Jede /mcp-Anfrage erfordert ein gültiges Bearer-Token oder einen API-Schlüssel.
Sitzungsentführungsabwehr: Sitzungs-IDs können nicht über verschiedene Identitäten hinweg geteilt werden.
Prompt-Injection-Abwehr: Regex- und heuristischer Analysator, der E-Mail-Felder mit Sicherheits-Audit-Logging scannt.
Empfänger-Whitelisting & Ratenbegrenzung: Verhindert unbefugten E-Mail-Abfluss und Überflutung.
Saubere Fehlerbereinigung: Generische, sichere JSON-RPC-Fehler (keine Stack-Trace-Lecks).
Mock-Tools:
get_current_datetime(timezone?): Systemdatum/-uhrzeit mit IANA-Zeitzonenvalidierung.
send_email(to, subject, body): Server-SMTP mit sicherer Simulations-Demo-Modus.
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 (optional, für MCP Inspector)
pip install -r requirements.txtpython -m src.serverSie werden sehen:
{"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_datetimeBeschreibung: Gibt das aktuelle Host-Systemdatum und die aktuelle Uhrzeit mit optionaler IANA-Zeitzonenkonvertierung zurück.
Parameter:
{
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Optional IANA timezone name (e.g. 'UTC', 'America/New_York', 'Asia/Kolkata')."
}
},
"required": []
}send_emailBeschreibung: Sendet eine E-Mail-Benachrichtigung über Server-SMTP mit strengen Sicherheitsschutzmechanismen.
Parameter:
{
"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/inspectorVerbinden mit URL: http://localhost:8100/mcp mit benutzerdefiniertem Header: Authorization: Bearer vishal-test-token.
1. get_current_datetime aufrufen:
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 aufrufen:
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."
}
}
}'Um den MCP-Server für eine öffentliche URL / Cloud-Demo freizugeben:
# 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 8100Führen Sie den automatisierten Testrunner aus, um alle 6 Demo-Fälle mit einem Befehl zu überprüfen:
python scripts/test_demo_cases.pypytest -vFall 1: Autorisierter Aufruf jedes Tools ist erfolgreich (sowohl get_current_datetime als auch send_email).
Fall 2: Nicht autorisierter Aufrufer erhält einen generischen Fehler (HTTP 401, Fehlercode -32001).
Fall 3: Fehlerhafte Eingabe gibt einen sauberen Validierungsfehler zurück (ungültige Zeitzone oder fehlende Schemafelder).
Fall 4: Prompt-Injection-Versuche in E-Mails werden blockiert und protokolliert (z. B. System-Prompt-Überschreibungen).
Fall 5: Senden an nicht auf der Whitelist stehende E-Mail wird abgelehnt (blockiert externe/unbefugte Adressen).
Fall 6: Wiederverwendung der Sitzungs-ID über verschiedene Identitäten hinweg wird verweigert (HTTP 403, Fehlercode -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
