Factory Intelligence MCP Server

This is a production-ready MCP (Model Context Protocol) server providing KPI tools for a Factory Intelligence dashboard. It communicates via the Stdio transport and leverages TimescaleDB for efficient time-series analysis, calculating Productivity, Quality, Downtime metrics, and diagnosing Alarms.

Features

1. Productivity KPI (get_productivity_kpi): Computes production efficiency against targets.

2. Quality KPI (get_quality_kpi): Calculates Yield % and Defect Rate %.

3. Downtime KPI (get_downtime_kpi): Analyzes machine availability based on production gaps.

4. KPI Summary (get_kpi_summary): Bundles all metrics for high-level dashboards.

5. Downtime Alarms Analysis (get_downtime_alarms_analysis): Correlates alarms with downtime periods to identify root causes.

Related MCP server: MCP SQL Server

Setup & Installation

Prerequisites

• Python 3.10+

• uv (recommended) or pip

• A running PostgreSQL/TimescaleDB instance with the factory schema.

1. Installation

git clone https://github.com/lvshrd/Factory-Intelligence-MCP-Server.git
cd Factory-Intelligence-MCP-Server
uv sync  # Installs dependencies including mcp, psycopg2, python-dateutil

2. Configuration

The server requires a DATABASE_URL environment variable. You have two options:

Option A: .env file (Recommended for local dev) Create a .env file in the Factory-Intelligence-MCP-Server directory:

DATABASE_URL="postgresql://username:password@localhost:5432/ProductionDB"

Option B: Environment Variable Injection Pass the DATABASE_URL directly through your MCP client configuration (see below).

Integration Guide

1. Using with Claude Desktop / Cursor

You can configure this server in Claude Desktop or Cursor's MCP settings.

Add this to your claude_desktop_config.json (or Cursor's MCP settings):

{
  "mcpServers": {
    "factory-intelligence": {
      "command": "uv",
      "args": [
        "--directory",
        "/ABSOLUTE/PATH/TO/Factory-Intelligence-MCP-Server", 
        "run",
        "server.py"
      ],
      "env": {
        "DATABASE_URL": "postgresql://username:password@localhost:5432/ProductionDB"
      }
    }
  }
}

2. Using with LangGraph / LangChain (Python)

To integrate this server programmatically using the official LangChain MCP client:

from langchain_mcp_adapters.client import MultiServerMCPClient

# Initialize client with Stdio transport
client = MultiServerMCPClient(
    {
        "factory-intelligence": {
            "transport": "stdio",
            "command": "uv",
            "args": [
                "--directory", 
                "/ABSOLUTE/PATH/TO/Factory-Intelligence-MCP-Server", 
                "run", 
                "server.py"
            ],
            "env": {
                "DATABASE_URL": "postgresql://username:password@localhost:5432/ProductionDB"
            }
        }
    }
)

Tool Definitions & Schemas

All tools share a common input structure requiring start_time and end_time.

1. get_productivity_kpi

Computes productivity metrics based on total good and bad bottles produced versus a target.

• Inputs:

  • start_time (string, ISO 8601)

  • end_time (string, ISO 8601)

• Outputs:

  • summary: Object containing value (ratio), total_production, good_count, bad_count.

  • timeseries: Array of { timestamp, value } objects.

  • metadata: Info on data source and computation notes.

2. get_quality_kpi

Computes Quality (Yield %) and Defect Rate %.

• Inputs: start_time, end_time (ISO 8601)

• Outputs:

  • summary: yield_percentage, defect_rate_percentage.

  • timeseries: Trend of Yield % over time.

3. get_downtime_kpi

Calculates uptime and downtime duration based on production gaps (zero production intervals).

• Inputs: start_time, end_time (ISO 8601)

• Outputs:

  • summary: uptime_seconds, downtime_seconds, availability_percentage.

4. get_kpi_summary

Bundles Productivity, Quality, and Downtime KPIs into a single response.

• Inputs: start_time, end_time (ISO 8601)

• Outputs:

  • productivity: Summary object from Tool 1.

  • quality: Summary object from Tool 2.

  • downtime: Summary object from Tool 3.

5. get_downtime_alarms_analysis

Identifies and ranks alarms that were active during inferred downtime periods.

• Inputs: start_time, end_time (ISO 8601)

• Outputs:

  • summary: Total downtime events and top alarm count.

  • top_alarms: List of alarms with frequency and total_duration_during_downtime.

  • downtime_events_sample: List of specific downtime windows (start, end, duration).

Example Tool Calls & Outputs

AI Agent Usage Example

Below is a demonstration of an AI agent (Cursor) calling the tools to analyze productivity and downtime root causes:

Request (Client -> Server)

Calling get_productivity_kpi for a single day:

{
  "name": "get_productivity_kpi",
  "arguments": {
    "start_time": "2025-12-10T00:00:00Z",
    "end_time": "2025-12-10T23:59:59Z"
  }
}

Response (Server -> Client)

Note: The result field contains the actual tool payload.

{
"tool": "get_productivity_kpi",
"inputs": {
"start_time": "2025-12-10T00:00:00Z",
"end_time": "2025-12-10T23:59:59Z"
},
"result": {
"summary": {
"kpi_name": "Productivity",
"value": 0.2019,
"total_production": 54074.0,
"good_count": 53473.0,
"bad_count": 601.0,
"unit": "ratio"
},
"timeseries": [
{
"timestamp": "2025-12-10T00:00:00+00:00",
"value": 54074.0
}
],
"metadata": {
"data_source": "agg_counter_1hour",
"bucket_width": "1 day",
"computation_note": "Target based on max observed speed (11160 BPH)"
}
},
"status": "ok",
"errors": []
}

Engineering Design Notes

1. Why specific tables were used?

• agg_counter_10sec_delta (The Source of Truth): Used for precise logic like Downtime Inference. Its delta-based structure allows us to accurately determine "zero production" intervals at a high resolution (10 seconds).

• agg_counter_1min / agg_counter_1hour (Performance): Used for KPI calculations over longer time ranges. Querying pre-aggregated data reduces the number of rows scanned by orders of magnitude (e.g., 1 year of 1-hour data is ~8,760 rows, vs ~3.1 million rows for 10-second data).

• agg_boolean_state_durations: Used for Alarm analysis because it natively stores state intervals (start, end, value), making overlap queries significantly easier than reconstructing states from raw timeseries events.

2. Assumptions Made

• Downtime Inference: We assume Zero Production = Downtime. Any 10-second bucket with sum(delta) = 0 is treated as a stop.

• Target Production: Calculated dynamically using a "Design Speed" of 11,160 Bottles Per Hour. This rate was derived from analyzing the historical data to find the maximum observed production in a single 10-second interval (31 bottles), ensuring the productivity ratio is relative to the machine's demonstrated peak capacity.

• Alarm Correlation: We assume that if an alarm is active (value=true) and its time interval overlaps with a downtime event, it is related to that downtime.

3. Performance Considerations

• Dynamic Aggregation Strategy: The system implements an intelligent router (get_aggregation_strategy) that selects the optimal table based on query duration:

  • < 10 mins -> agg_counter_1min (High detail)

  • < 30 mins -> agg_counter_30min (Medium detail)

  • < 12 hours -> agg_counter_1hour (Balanced)

  • > 12 hours -> agg_counter_1hour (Aggregated to Daily buckets on-the-fly)

• SQL-Side Computation: Heavy logic is pushed to the database.

  • Downtime: Instead of fetching millions of rows to Python, we use SQL CTEs and COUNT(*) FILTER to calculate uptime/downtime seconds instantly.

  • Alarm Analysis: We use "Gaps and Islands" logic (using ROW_NUMBER()) inside the database to merge continuous zero-production buckets into downtime events, preventing data explosion in the application layer.

Testing

Run the included verification script to see all tools in action:

uv run test_kpi_service.py