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

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

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

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

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:

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

{
  "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.