# Industrial Motor Telemetry — MCP Server (PoC)

A Proof of Concept that exposes an **industrial motor telemetry system** to an
AI agent through the **Model Context Protocol (MCP)**, while also serving a
conventional **FastAPI REST** interface — both backed by the *same* async
service layer.

## Why this design

| Concern | Where it lives | Rationale |
|---|---|---|
| Data contract | `models.py` (`MotorStatus`, `MotorHealth`) | Single source of truth, validated by Pydantic v2. |
| Business logic | `telemetry_service.py` (`get_motor_status`) | Transport-agnostic async logic; no FastAPI/MCP imports → unit-testable and reusable. |
| Transports | `main.py` | MCP tool **and** REST endpoint both delegate to the one service coroutine — zero logic duplication. |

The MCP server (built on the official `mcp` SDK's `MCPServer`) is served over
**Streamable HTTP** and mounted onto the FastAPI app, so a single `uvicorn`
process exposes both protocols.

## Requirements

- Python 3.10+
- Dependencies in `requirements.txt` (FastAPI, Uvicorn, Pydantic v2, `mcp` v2)

## Run locally

```bash
cd motor-telemetry-mcp
python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -r requirements.txt

# Start the combined REST + MCP server
uvicorn main:app --reload --port 8000
```

- MCP endpoint (Streamable HTTP): `http://localhost:8000/mcp`
- REST docs (Swagger UI): `http://localhost:8000/docs`

## Try the REST API

```bash
curl http://localhost:8000/motors                 # list motor IDs
curl http://localhost:8000/motors/MTR-001         # NORMAL
curl http://localhost:8000/motors/MTR-003         # CRITICAL (overheating)
curl -i http://localhost:8000/motors/MTR-999      # 404 with clear detail
```

## Run the tests

```bash
pip install -r requirements-dev.txt
pytest
```

The suite (14 tests) covers all three layers with no network sockets:

- `tests/test_service.py` — async domain logic (health mapping, forgiving
  input normalization, not-found error).
- `tests/test_rest_api.py` — FastAPI endpoints via httpx's in-process ASGI
  transport (200s, structured body, 404).
- `tests/test_mcp_tool.py` — MCP layer in-process: tool discovery, input
  schema, structured output, and the tool-error path.

## Run with Docker

```bash
docker build -t motor-telemetry-mcp .
docker run --rm -p 8000:8000 motor-telemetry-mcp
```

The image is slim, installs deps in a cached layer, runs as a **non-root**
user, and ships a container `HEALTHCHECK` against `/health`.

## Mock fleet

| Motor ID | Temp (°C) | Vibration (mm/s) | RPM | Status |
|---|---|---|---|---|
| `MTR-001` | 62.5 | 1.8 | 1490 | `NORMAL` |
| `MTR-002` | 83.0 | 4.2 | 1475 | `WARNING` |
| `MTR-003` | 118.4 | 11.6 | 1360 | `CRITICAL` |

## How an AI agent connects to the MCP tool

The server advertises one tool, `get_motor_status_tool`, and one resource,
`motor://fleet`. An MCP-capable agent runtime discovers and invokes it exactly
like the included `client_example.py`:

```bash
# In a second terminal, with the server running:
python client_example.py
```

Conceptual flow:

1. **Connect** — the agent's MCP client opens a Streamable-HTTP session to
   `http://localhost:8000/mcp`.
2. **Discover** — it calls `list_tools()` and receives the schema for
   `get_motor_status_tool` (auto-generated from the Python type hints).
3. **Invoke** — when the user asks *"Is motor MTR-003 healthy?"*, the LLM emits
   a tool call `get_motor_status_tool(motor_id="MTR-003")`.
4. **Reason** — the server returns structured JSON (temperature, vibration,
   rpm, status). The agent reads `status: "CRITICAL"` and can respond /
   escalate. Unknown motors return an MCP tool *error* result, which the agent
   can surface gracefully.

### Wiring it into a real client (e.g. Claude Desktop / any MCP host)

Point the host at the Streamable-HTTP URL. Example host config entry:

```json
{
  "mcpServers": {
    "motor-telemetry": {
      "type": "http",
      "url": "http://localhost:8000/mcp"
    }
  }
}
```

## Project layout

```
motor-telemetry-mcp/
├── models.py             # Pydantic models: MotorStatus, MotorHealth enum
├── telemetry_service.py  # Async, transport-agnostic domain logic + mock data
├── main.py               # FastAPI app + MCP server (Streamable HTTP) mounted
├── client_example.py     # Demo MCP client that calls the tool
├── tests/                # pytest suite (service, REST, MCP layers)
├── Dockerfile            # Slim, non-root container image
├── .dockerignore
├── pytest.ini
├── requirements.txt
├── requirements-dev.txt  # test/dev dependencies
└── README.md
```

## Production notes (talking points)

- Swap `telemetry_service` internals for a real historian / OPC-UA / time-series
  source — no transport code changes needed.
- `stateless_http=True` keeps the MCP transport horizontally scalable.
- Add auth (the `mcp` SDK supports token verifiers / OAuth resource servers) and
  rate limiting before exposing beyond localhost.
```