Survey Generation & Email Workflow

Project Scope

This repository implements a queued survey generation workflow for PDF documents received by Gmail. It uses ActiveMQ as the message broker, PostgreSQL and MongoDB for persistence, and Google Gemini for natural language processing. The workflow extracts text from incoming PDFs, generates a structured survey summary, produces HTML or PDF output, and sends the result back to the original sender.

Related MCP server: Gmail MCP Server

What this project does

• Reads unread Gmail messages and downloads attached PDF files

• Extracts text from PDFs using PyMuPDF

• Queues document processing tasks with ActiveMQ / STOMP

• Consumes tasks in a worker process

• Generates report summaries using Google Gemini (google.genai preferred)

• Converts generated summaries into HTML email content

• Sends HTML email replies via Gmail API

• Tracks metadata in PostgreSQL and task status in MongoDB

• Uses environment-based configuration via .env

Libraries and Technologies Used

• python-dotenv - load environment variables from .env

• stomp.py - STOMP client for ActiveMQ

• psycopg2-binary - PostgreSQL database adapter

• pymongo - MongoDB client

• google-genai / google-generativeai - Gemini text generation

• google-api-python-client / google-auth-oauthlib / google-auth-httplib2 - Gmail API integration

• PyMuPDF - PDF text extraction

• reportlab - PDF generation from survey text

• markdown - Markdown-to-HTML conversion

• opentelemetry-api / opentelemetry-sdk / opentelemetry-exporter-otlp - telemetry and tracing

• docker - project includes Docker Compose support for service orchestration

• Grafana - dashboards and provisioning for monitoring and visualization

High-level Flow

1. Email ingestion

   • app/server.py polls Gmail for unread messages and downloads PDF attachments via app/gmail_utils.py

   • Each valid PDF becomes a task message published to ActiveMQ

   • Metadata is stored in PostgreSQL and MongoDB as pending

2. Task queueing

   • app/queue_manager.py publishes the task to the configured ActiveMQ queue

   • Each task includes email_id, sender, pdf_path, document_uuid, and document_name

3. Worker consumption

   • app/worker.py subscribes to the queue and receives tasks

   • When a task arrives, the worker:

     • marks the task in_progress in MongoDB

     • extracts PDF text via app/pdf_utils.py

     • generates survey content with app/gemini_utils.py

     • converts the survey into HTML via app/survey_utils.py

     • sends the HTML email using app/gmail_utils.py

     • updates document status in PostgreSQL and MongoDB

4. Output and persistence

   • PDF text extraction, survey generation, emailing, and state updates are logged

   • uploads/ stores downloaded email attachments

   • outputs/ stores generated survey PDFs

Directory Structure

MCP/
├── .env                # Local configuration values (private)
├── .env.example        # Example environment variables
├── docker-compose.yml  # Service orchestration template
├── otel-collector-config.yaml
├── requirements.txt    # Python dependencies
├── app/
│   ├── config.py       # dotenv loader
│   ├── gemini_utils.py # Gemini model invocation and text extraction
│   ├── gmail_utils.py  # Gmail API helpers for download/send
│   ├── mongodb_db.py   # MongoDB task status store
│   ├── pdf_utils.py    # PDF text extraction via PyMuPDF
│   ├── postgres_db.py  # PostgreSQL metadata store
│   ├── queue_manager.py# ActiveMQ STOMP connection, publish, subscribe
│   ├── server.py       # Queue producer / FastMCP tool wrapper
│   ├── survey_utils.py # HTML/PDF generation from survey text
│   ├── telemetry.py    # Logging and tracing helpers
│   ├── test_gmail.py   # Gmail-related tests/examples
│   └── worker.py       # Queue consumer and task processor
├── credentials/        # Gmail OAuth credentials
├── outputs/            # Generated survey output files
├── grafana/
│   ├── dashboards/
│   │   └── survey_dashboard.json
│   └── provisioning/
│       ├── dashboards/
│       │   └── dashboard.yml
│       └── datasources/
│           └── postgres.yml
└── uploads/            # Downloaded PDF attachments

Core Components

app/server.py

• Exposes FastMCP tools for document processing and inbox workflow

• Primary workflow is process_inbox()

• Reads unread Gmail messages

• Downloads PDF attachments and stores them locally

• Writes metadata to PostgreSQL

• Writes pending task status to MongoDB

• Publishes processing tasks to ActiveMQ

app/worker.py

• Runs the background consumer

• Connects to ActiveMQ, PostgreSQL, and MongoDB

• Subscribes to the configured queue

• Processes each message by:

  • extracting PDF text

  • generating survey text

  • producing HTML content

  • emailing the result

  • updating both databases with final status

app/queue_manager.py

• Handles ActiveMQ connectivity via STOMP

• Validates broker protocol on startup

• Supports infinite connection timeout with ACTIVEMQ_CONNECT_TIMEOUT=0

• Allows publish and subscribe operations

• Disabled STOMP heartbeats for long-running worker tasks

app/gemini_utils.py

• Uses google.genai when available

• Falls back to google.generativeai for compatibility

• Sends a structured prompt to Gemini

• Extracts text safely from the response

app/gmail_utils.py

• Uses the Gmail API for:

  • fetching unread messages

  • downloading attachment payloads

  • parsing sender email addresses

  • sending HTML-rich emails

• Stores attachments under uploads/

• Builds clean email content with fallback text formatting

app/pdf_utils.py

• Extracts text from PDF documents using PyMuPDF

• Used as the first processing step in the worker

app/survey_utils.py

• Generates PDF output from survey text via ReportLab

• Converts survey text to styled HTML for email delivery

• Includes a fallback HTML generator for plain text

app/postgres_db.py

• Connects to PostgreSQL

• Creates a documents table if missing

• Stores sender, document UUID, file name, and final status

• Tracks result status for each processed document

app/mongodb_db.py

• Connects to MongoDB

• Stores and updates task status documents

• Supports index creation for document_uuid, sender, and timestamps

app/config.py

• Loads .env values with python-dotenv

• Centralizes environment configuration for the whole app

app/telemetry.py

• Simple console logging helper

• Provides OpenTelemetry span support for tracing operations

Grafana provisioning & dashboards

• Dashboard JSON: grafana/dashboards/survey_dashboard.json contains pre-built panels for survey processing metrics and document status.

• Provisioning files: grafana/provisioning/dashboards/dashboard.yml and grafana/provisioning/datasources/postgres.yml allow Grafana to auto-provision the dashboard and a PostgreSQL datasource on startup.

• Ensure the datasource configuration matches your POSTGRES_* settings so dashboard panels can query persisted metadata.

Setup

1. Create and activate your Python virtual environment

python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

2. Copy environment example

cp .env.example .env

3. Populate .env

• GEMINI_API_KEY or GOOGLE_API_KEY

• ACTIVEMQ_HOST, ACTIVEMQ_PORT, ACTIVEMQ_USER, ACTIVEMQ_PASSWORD

• POSTGRES_HOST, POSTGRES_PORT, POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_DB

• MONGODB_URI, MONGODB_DB, MONGODB_COLLECTION

4. Provide Gmail OAuth credentials in credentials/credentials.json

5. Start ActiveMQ, PostgreSQL, and MongoDB services

Running the project

Start the worker

python app/worker.py

Trigger inbox processing

• The inbox workflow is exposed via app/server.py as a FastMCP tool

• In practice, run the tool path or call process_inbox() from your FastMCP interface

• Alternatively, you can add a small script to invoke process_inbox() directly

Start Grafana (optional)

To preview processing metrics and the included dashboard locally you can run Grafana and mount the provisioning files. Example using docker run:

docker run -d -p 3000:3000 \
   -v "$(pwd)/grafana/provisioning:/etc/grafana/provisioning" \
   -v "$(pwd)/grafana/dashboards:/var/lib/grafana/dashboards" \
   --name grafana grafana/grafana:latest

If you use docker-compose, mount the grafana/ directory into your Grafana service under /etc/grafana/provisioning so Grafana will auto-provision the dashboard on startup.

Important Notes

• uploads/ stores downloaded PDF attachments from Gmail

• outputs/ stores generated survey PDF files

• ACTIVEMQ_CONNECT_TIMEOUT=0 disables the STOMP connection timeout for long-running workers

• Gmail credentials and API tokens must remain private and should not be committed

Extended Notes

• The worker is designed to stay alive while processing long Gemini requests

• The queue consumer logs disconnects and reconnects if the broker drops connection

• PostgreSQL stores stable document metadata, while MongoDB stores live task status changes

• The project is structured to separate email ingestion, queue orchestration, NLP generation, and output delivery cleanly

Future Improvements

• Add a scheduler or webhook trigger for process_inbox()

• Add retry logic for failed tasks and emails

• Add unit tests around queue, DB, and Gemini flows

• Add a health-check endpoint or monitoring integration

Grafana dashboard

• A pre-built Grafana dashboard is included at grafana/dashboards/survey_dashboard.json.

• Grafana provisioning files are available in grafana/provisioning/dashboards/dashboard.yml and grafana/provisioning/datasources/postgres.yml.

• To enable the dashboard, mount the grafana/ directory into your Grafana container's provisioning path or copy the provisioning files into your Grafana server; Grafana will auto-provision the dashboard on startup when provisioning is enabled.

• Ensure the datasource configuration points to your PostgreSQL instance used by this project so panels can query the stored metadata and metrics.