DataFlow MCP Server - Production Grade

A secure, production-ready Model Context Protocol (MCP) server with MongoDB integration, featuring comprehensive security controls, CRUD operations, logging, and monitoring.

📋 Features

Security

• ✅ Input Validation & Sanitization - Prevents NoSQL injection attacks

• ✅ MongoDB SSL/TLS Support - Secure cloud deployments

• ✅ Rate Limiting - Protects against abuse (100 req/min default)

• ✅ Connection Pooling - Optimized for performance

• ✅ Document Size Limits - Prevents resource exhaustion

• ✅ Field Name Validation - Blacklists dangerous operators

Operations

• ✅ CRUD Operations - Create, Read, Update, Delete documents

• ✅ Filtering & Pagination - Flexible data retrieval with limits

• ✅ Sorting Support - Sort by any field (ascending/descending)

• ✅ Bulk Operations Ready - Extensible architecture

Monitoring & Observability

• ✅ Comprehensive Logging - File & console with rotation

• ✅ Health Checks - Service health status endpoint

• ✅ Metrics Tracking - Request counts, success rates

• ✅ Error Handling - Detailed error reporting

Production Ready

• ✅ Security First - SSL/TLS support, input validation

• ✅ Environment Config - 12-factor app ready

• ✅ Graceful Shutdown - Proper resource cleanup

Related MCP server: Kroki MCP

🚀 Quick Start

Prerequisites

• Python 3.12+

• Docker & Docker Compose (optional)

• MongoDB (or use Docker Compose)

Local Development

1. Clone and setup:

cd dataflow_mcp
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -e .

2. Configure environment:

cp .env.example .env
# Edit .env with your MongoDB connection

3. Run the server:

python main.py

📡 API Tools

Health Check

Get server status and metrics.

{
  "status": "healthy",
  "uptime_seconds": 123.45,
  "metrics": {
    "total_requests": 42,
    "successful_requests": 40,
    "failed_requests": 2,
    "success_rate": 95.24
  }
}

Read Collection

Retrieve documents with filtering, pagination, and sorting.

Parameters:

• collection_name (required): Collection name

• filter_query: JSON string with MongoDB filter

• limit: Max documents (default: 100, max: 1000)

• skip: Skip N documents (default: 0)

• sort_by: Field to sort by

Example:

{
  "collection_name": "users",
  "filter_query": "{\"status\": \"active\"}",
  "limit": 10,
  "skip": 0,
  "sort_by": "created_at"
}

Get Document

Retrieve a single document by ID.

Parameters:

• collection_name: Collection name

• document_id: MongoDB ObjectId as string

Create Document

Create a new document in a collection.

Parameters:

• collection_name: Collection name

• document_json: JSON string representing the document

Example:

{
  "collection_name": "users",
  "document_json": "{\"name\": \"John\", \"email\": \"john@example.com\", \"status\": \"active\"}"
}

Update Document

Update an existing document.

Parameters:

• collection_name: Collection name

• document_id: MongoDB ObjectId as string

• update_json: JSON with fields to update

Example:

{
  "collection_name": "users",
  "document_id": "65f8a1b2c3d4e5f6g7h8i9j0",
  "update_json": "{\"status\": \"inactive\", \"updated_at\": \"2024-01-01T12:00:00Z\"}"
}

Delete Document

Delete a document from a collection.

Parameters:

• collection_name: Collection name

• document_id: MongoDB ObjectId as string

🔒 Security Features

Input Validation

• Collection names: Alphanumeric, dash, underscore only

• Field names: Prevents dangerous operators ($where, $function, etc.)

• Filters: Maximum 10KB, blacklist dangerous operations

• Documents: Maximum 1MB, enforced size limits

MongoDB Security

• Connection Options:

  • Connection pooling (default: 10 connections)

  • Retry writes enabled

  • Write concern: majority

  • Journaling enabled

  • SSL/TLS for cloud deployments

• Environment Variables:

  MONGO_USE_TLS=true
  MONGO_CA_CERT_PATH=/path/to/ca.pem
  MONGO_ALLOW_INVALID_CERTS=false

Rate Limiting

• 100 requests per 60 seconds (configurable)

• Per-client tracking

• Returns clear error on limit exceeded

Error Handling

• Safe error messages (no sensitive data leaks)

• Detailed internal logging

• Graceful degradation

📊 Environment Variables

Required

MONGO_URI=mongodb://user:password@host:port/database
MONGO_DB_NAME=dataflow

Optional (with defaults)

MONGO_TIMEOUT=5000              # Connection timeout (ms)
MONGO_POOL_SIZE=10              # Connection pool size
MONGO_MAX_IDLE_TIME=45000       # Max idle time (ms)
MONGO_USE_TLS=false             # Enable TLS
MONGO_CA_CERT_PATH=             # CA certificate path
LOGS_DIR=./logs                 # Log directory
LOG_LEVEL=INFO                  # Logging level

📁 Project Structure

dataflow_mcp/
├── core.py            # FastMCP instance, rate limiter, metrics, prompt loading, normalization
├── server.py          # tool registration + mcp.run()
├── tools/
│   ├── health.py      # health_check, database_status
│   ├── crud.py        # generic MongoDB CRUD tools
│   ├── images.py      # contest banner pipeline (missing/broken images, cover prompts)
│   ├── migration.py   # v4.0 schema migration/backfill tools
│   ├── contests.py    # structuring + full generation + detail generation
│   ├── events.py      # events pipeline (fetch → structure → submit → query)
│   ├── raw_data.py    # raw scraped data bridge + overview
│   ├── validation.py  # chatbot-driven web validation pipeline
│   └── audit.py       # duplicate audit + discrepancy flagging
config/
├── mongodb.py           # MongoDB connection with pooling
├── security.py          # Validation and rate limiting
└── logging_config.py    # Logging setup
tools/                   # service layer (DataManager, generators, dedup gate, validators)
prompts/                 # prompt files (descriptive names + Prompts*.txt aliases)
main.py                  # thin entry point → dataflow_mcp.server
├── pyproject.toml      # Dependencies and config (console script: dataflow-mcp)
└── .env.example        # Environment template

🎪 Events Pipeline

The MCP server includes a full events pipeline so AI chatbots can harvest and structure participatory events (conferences, summits, workshops, webinars, meetups, trainings, …):

1. get_records_for_events(source=..., limit=10)  → raw URLs + events-v1.1 prompt
2. [chatbot researches each URL and outputs event JSON]
3. submit_structured_events(events_json)         → persists to the Events collection
4. get_events(event_type=..., upcoming_only=true) → read structured events back
5. get_events_overview()                          → counts by type/status
6. get_events_for_detail_generation(batch_size=10) → events + event-details-v1.0.txt prompt
7. [chatbot researches and writes event details]
8. submit_event_details(event_id, details_json)  → versioned event_details saved
9. get_event_detail_status()                       → coverage metrics (remaining events to generate)

Event detail pages mirror the contest detail flow: EventDetailGenerator (tools/event_detail_generator.py) provides the priority queue, quality validation, and versioned event_details storage.

Prompt files were renamed to descriptive names (contest-structuring-v4.0.txt, event-structuring-v1.1.txt, …) with the old Prompts*.txt names kept as aliases. See TOOLS_REFERENCE.md for the full tool reference.

🔧 Configuration for Cloud Deployment

AWS Deployment

MONGO_URI=mongodb+srv://user:password@cluster.mongodb.net/dataflow
MONGO_USE_TLS=true
MONGO_ALLOW_INVALID_CERTS=false

Azure Deployment

MONGO_URI=mongodb://user:password@host.mongo.cosmos.azure.com:10255/database
MONGO_USE_TLS=true
MONGO_CA_CERT_PATH=/etc/ssl/certs/ca-certificates.crt

GCP Deployment

MONGO_URI=mongodb://user:password@instance:27017/database
MONGO_USE_TLS=true

🚨 Production Checklist

• MongoDB backups configured

• SSL/TLS certificates installed

• Environment variables set securely (not in code)

• Logs redirected to centralized logging

• Health checks configured in load balancer

• Rate limits adjusted for your use case

• MongoDB indexes optimized

• Connection pool size tuned

• Monitoring/alerting setup

• Graceful shutdown tested

📈 Performance Optimization

MongoDB Indexes

Pre-created indexes in scripts/mongo-init.js:

• User email: unique constraint

• Timestamps: for sorting and TTL

• Status: for filtering

Connection Pooling

• Default pool size: 10 (adjust via MONGO_POOL_SIZE)

• Min connections: 2 (automatically maintained)

• Max idle time: 45 seconds

Request Limits

• Max filter size: 10KB

• Max document size: 1MB

• Max page size: 1000 documents

• Rate limit: 100 req/min

🧪 Testing & Development

Install dev dependencies:

pip install -e ".[dev]"

Run tests:

pytest --cov=tools --cov=config

Code formatting:

black .
flake8 .
mypy .

📝 Logging

Logs are written to:

• File: ./logs/mcp_server_YYYYMMDD.log (rotated daily, max 10MB)

• Console: Real-time output

Log levels:

• DEBUG - Detailed diagnostic info

• INFO - General events

• WARNING - Warning messages

• ERROR - Error events

🐛 Troubleshooting

MongoDB Connection Failed

Check MONGO_URI and credentials
Verify MongoDB is running: mongosh "mongodb://..."
Check network connectivity and firewall

Rate Limit Exceeded

Default: 100 requests per 60 seconds
Increase MONGO_POOL_SIZE and optimize queries
Implement request queuing on client

High Memory Usage

Reduce MONGO_POOL_SIZE
Lower MONGO_MAX_IDLE_TIME
Check for large result sets (use pagination)

📚 References

• FastMCP Documentation

• MongoDB Security

• Connection String Format

• Docker Best Practices

📄 License

MIT License - See LICENSE file for details

👤 Support

For issues and questions:

1. Check troubleshooting section

2. Review logs in ./logs/

3. Check MongoDB connection

4. Verify environment variables

Built for production-grade data operations with security-first design.

dataflow_mcp