# 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.

---

## Setup & Installation

### Prerequisites
- Python 3.10+
- `uv` (recommended) or `pip`
- A running PostgreSQL/TimescaleDB instance with the factory schema.

### 1. Installation
```bash
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:
```env
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):
```json
{
  "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"
      }
    }
  }
}
```

<img src="assets/mcp%20server%20loaded.png" alt="MCP Server Loaded" width="500" class="center"/>

### 2. Using with LangGraph / LangChain (Python)
To integrate this server programmatically using the official LangChain MCP client:

```python
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:

<img src="assets/screenshot.png" alt="Agent Usage Demo" width="300"/>

### Request (Client -> Server)
Calling `get_productivity_kpi` for a single day:

```json
{
  "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.
```json
{
"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:
```bash
uv run test_kpi_service.py
```