# FastMCP IoT Sensor Server

A portfolio artifact demonstrating hands-on [MCP (Model Context Protocol)](https://modelcontextprotocol.io) server
development using [FastMCP](https://gofastmcp.com). Simulates a network of IoT environmental sensors — the kind of
fleet you'd find monitoring a building or vehicle for emergency-response / edge-monitoring use cases — and exposes
their data and controls as MCP tools that any MCP client (Claude Desktop, `fastmcp dev`, or a custom client) can
query and act on.

For the layered architecture and a dependency diagram see [`docs/architecture.md`](docs/architecture.md);
for a deeper file-by-file walkthrough (useful for talking through this project in an interview)
see [`docs/project-walkthrough.md`](docs/project-walkthrough.md); for the MCP protocol/design
background behind the decisions here see [`docs/mcp-design-notes.md`](docs/mcp-design-notes.md).

## Why this exists

This is a v1 portfolio demo, not a production system: no real hardware, no message broker, everything runs in a
single Python process. It's designed to show the shape of a real MCP server — tool design, resource vs. tool
distinction, background async work alongside request/response handling, and persistence — without the operational
overhead of an actual sensor deployment. It pairs naturally with a follow-up "real edge gateway" project (MQTT +
hardware) as a next step.

**The MCP server and the simulation are deliberately separate concerns.** `iot_sensors.server` (and everything it
depends on — `app.py`, `domain/`, `tools/`, `resources.py`, `prompts.py`) is the thing another project would
actually embed or connect to; it never imports simulation code. `iot_sensors.simulation` exists purely to
generate the live, changing data that makes the server interesting to demo — it's an optional add-on, off by
default (see [Simulating live data](#simulating-live-data) below), not part of the server's own runtime.

## How it works

Two things run as **separate OS processes** and never talk to each other directly — they
communicate only indirectly, through the shared SQLite file. This is deliberate: the MCP server
doesn't need to know a simulator exists, and the simulator doesn't need to know anything about
MCP.

```mermaid
graph LR
    Client["MCP Client<br/>(Claude Desktop, Inspector,<br/>or a custom client)"]

    subgraph SP["Server process — fastmcp run -m iot_sensors.server"]
        Transport["stdio transport<br/>(JSON-RPC)"]
        App["app.py<br/>(FastMCP instance + lifespan)"]
        Surface["tools/ · resources.py · prompts.py"]
    end

    subgraph SIM["Simulator process — python -m iot_sensors.simulation.run"]
        Simulator["simulator.py<br/>(asyncio loop, ~every 8s)"]
    end

    DB[("iot_sensors.db<br/>(SQLite — domain/db.py)")]

    Client <-->|"tool calls, resource reads,<br/>prompt requests"| Transport
    Transport <--> App
    App --> Surface
    Surface <-->|"query / write via domain.db"| DB
    Simulator -->|"writes readings + alerts<br/>via domain.db"| DB

    style DB fill:#f2e9d8,stroke:#a67c3a
    style SP fill:#e8eef7,stroke:#3a5a8c
    style SIM fill:#f7f0e8,stroke:#a67c3a
```

**How a single tool call actually flows through the MCP layer**, e.g. a client calling
`get_active_alerts`:

```mermaid
sequenceDiagram
    participant C as MCP Client
    participant T as stdio transport
    participant S as FastMCP (app.py)
    participant Tool as tools/alerts.py
    participant DB as domain/db.py (SQLite)

    C->>T: call_tool("get_active_alerts", {})
    T->>S: dispatch to registered tool
    S->>Tool: get_active_alerts()
    Tool->>DB: SELECT * FROM alerts WHERE acknowledged = 0
    DB-->>Tool: rows
    Tool-->>S: list[AlertRecord] (Pydantic)
    S-->>T: serialized structured result
    T-->>C: JSON-RPC response
```

FastMCP handles the schema generation, dispatch, and serialization steps (`S` above) — the tool
function itself only ever deals with plain domain objects, never the wire format. See
[`docs/architecture.md`](docs/architecture.md) for the static import/dependency diagram (which
files depend on which), as opposed to this runtime data-flow view.

## Architecture

The package is organized in layers rather than one flat pile of files: a **domain layer**
(`domain/`) that describes sensors/readings/alerts and how they're persisted, with no knowledge of
MCP or the simulator; an **MCP layer** (`app.py`, `server.py`, `tools/`, `resources.py`,
`prompts.py`) that exposes the domain layer over the MCP protocol; and a **simulation layer**
(`simulation/`) that also depends on the domain layer, but only to generate demo data — it has no
relationship to the MCP layer. Dependencies point one direction, inward: both the MCP layer and
the simulation layer import from `domain/`, but `domain/` imports from neither.

```
fastmcp.json                    Declarative FastMCP config: entrypoint, transport, deployment env
pyproject.toml                  Package metadata; src-layout package discovery
src/iot_sensors/
├── app.py                      Creates the shared FastMCP() instance + lifespan (DB init + seed; simulator only if opted in)
├── server.py                   Entrypoint: imports tool/resource/prompt modules for registration, then mcp.run()
├── domain/                     sensors/readings/alerts + persistence — knows nothing about MCP or simulation
│   ├── db.py                    SQLite schema (stdlib sqlite3, no ORM) + query helpers; DB path resolved next to this file
│   ├── models.py                Pydantic models for MCP tool/resource I/O, plus domain dataclasses
│   └── seed_data.py             Fleet identity: 9 sensors across 3 zones + default alert thresholds
├── resources.py                sensors://fleet resource
├── prompts.py                  diagnose_alert prompt
├── tools/
│   ├── sensors.py               list_sensors, get_sensor_reading, get_sensor_history
│   ├── alerts.py                set_alert_threshold, get_active_alerts, acknowledge_alert, bulk_acknowledge_alerts
│   └── zone_health.py           summarize_zone_health (sampling)
└── simulation/                 optional: not imported by server.py/app.py unless opted in
    ├── profiles.py               per-sensor drift/anomaly parameters — simulation-only, domain layer never reads this
    ├── simulator.py              asyncio background loop: readings with drift + occasional anomaly spikes
    └── run.py                    standalone script: run this in a separate process for a live demo
tests/                          pytest + pytest-asyncio tests that call tools/resources/prompts through an in-memory FastMCP Client
docs/                           architecture diagram, project walkthrough, MCP design notes
```

The MCP-layer structure follows the standard FastMCP src-layout convention: a thin `server.py`
entrypoint that just imports the tool/resource/prompt modules for their `@mcp.tool`/
`@mcp.resource`/`@mcp.prompt` registration side effects, with the actual `FastMCP()` instance
factored into its own `app.py` so those modules can import it without circular imports. Tools are
grouped into `tools/<domain-area>.py` files rather than piling every tool into one file — the
natural next step if the tool count kept growing would be splitting each domain area into its own
mounted sub-server (see `docs/mcp-design-notes.md`).

**Design rationale:**
- **SQLite over an ORM** — the schema is small and stable; raw `sqlite3` keeps the data layer legible without an
  abstraction that isn't earning its keep.
- **DB path resolved relative to the package, not the process's working directory** — `domain/db.py` derives the
  default path from its own file location (`Path(__file__).resolve().parent`), so the database always lands in the
  same place no matter where `fastmcp run` happens to be launched from. Override with `IOT_SENSORS_DB_PATH` if
  needed.
- **Simulation kept out of the server's own runtime** — the server's lifespan only initializes the schema and seeds
  the fleet; it starts the background simulator *only* if `IOT_SENSORS_ENABLE_SIMULATOR` is set. A project
  embedding `iot_sensors.server` gets a clean server with zero simulation code in its default startup path; the
  simulator is there to demo the server working, not to be a dependency of using it. This split runs down to the
  data too: `domain/seed_data.py` (core, always loaded) only holds a sensor's identity — id/name/type/zone/unit —
  and its default alert thresholds; `simulation/profiles.py` (simulation-only) holds the drift/anomaly parameters
  that only
  the simulator needs. Nothing outside `iot_sensors.simulation` ever reads `profiles.py`.
- **Tools vs. resources** — the eight tools are the actionable surface (query specific sensors, configure
  thresholds, acknowledge alerts, summarize zone health). The `sensors://fleet` resource demonstrates the
  complementary read-only, subscribable surface MCP offers for "give me the current state of the world" access.

## Sensor Fleet

| Zone   | Sensors                                  |
|--------|-------------------------------------------|
| Zone A | Temperature, Humidity, CO2                |
| Zone B | Temperature, CO2, Smoke/particulate index |
| Zone C | Temperature, Humidity, Smoke/particulate index |

When the simulator is running (see below), each sensor emits a new reading every ~8 seconds with small randomized
drift, plus a small chance per tick of an injected anomaly spike (useful for demoing the alerting path without
waiting for a real incident).

## Setup

Requires Python 3.11+.

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```

## Quick start

1. **Run the tests** (fast, deterministic, doesn't need the simulator):
   ```bash
   pytest tests/ -v
   ```
2. **Start the server** in one terminal:
   ```bash
   FASTMCP_SHOW_SERVER_BANNER=false fastmcp run -m iot_sensors.server
   ```
3. **Start the simulator** in a second terminal, so there's live data to query — the server alone
   only seeds the static fleet, it doesn't generate readings itself (see
   [Simulating live data](#simulating-live-data)):
   ```bash
   python -m iot_sensors.simulation.run
   ```
4. **Interact with it** via the [Inspector](#interactive-inspector) (instead of step 2 — the
   Inspector launches the server for you), Claude Desktop (see
   [below](#connect-from-claude-desktop)), or any other MCP client.

### Run the server standalone (stdio transport)

```bash
fastmcp run   # picks up fastmcp.json in the current directory
```

or, without the config file:

```bash
FASTMCP_SHOW_SERVER_BANNER=false fastmcp run -m iot_sensors.server
```

(`iot_sensors/server.py` uses relative imports, so it must be run as an installed module — `-m
iot_sensors.server` — rather than as a bare file path.)

### Interactive inspector

```bash
FASTMCP_SHOW_SERVER_BANNER=false npx @modelcontextprotocol/inspector@latest .venv/bin/fastmcp run -m iot_sensors.server
```

Use Inspector v2+ (`@latest`) — v1 doesn't implement a UI for elicitation requests, which
`bulk_acknowledge_alerts` relies on. Suppressing the startup banner matters for stdio transport:
any extra text on stdout besides JSON-RPC corrupts the message framing and the Inspector reports
the connection as failed.

### Simulating live data

The server on its own only initializes the schema and seeds the static fleet — every sensor's `last_reading` stays
`None` and there's nothing to query in `get_active_alerts()` until something writes readings. To see it behave like
a live system, either:

**Run the simulator in a separate process** (recommended — keeps the server itself simulation-free):
```bash
python -m iot_sensors.simulation.run
```

**Or opt the server's own lifespan into starting it:**
```bash
IOT_SENSORS_ENABLE_SIMULATOR=true FASTMCP_SHOW_SERVER_BANNER=false fastmcp run -m iot_sensors.server
```

### Connect from Claude Desktop

Add to your Claude Desktop MCP config (`claude_desktop_config.json`):

```json
{
  "mcpServers": {
    "iot-sensors": {
      "command": "/absolute/path/to/.venv/bin/fastmcp",
      "args": ["run", "-m", "iot_sensors.server"],
      "env": { "FASTMCP_SHOW_SERVER_BANNER": "false" }
    }
  }
}
```

Restart Claude Desktop and the `iot-sensors` server should appear in the MCP tool list.

### Run tests

```bash
pytest tests/ -v
```

## MCP Tools

| Tool | Description |
|------|-------------|
| `list_sensors()` | All sensors with current status (id, name, type, zone, last value, last updated) |
| `get_sensor_reading(sensor_id)` | Latest reading for one sensor |
| `get_sensor_history(sensor_id, since_minutes)` | Time-series of past readings |
| `set_alert_threshold(sensor_id, min, max)` | Configure alert bounds for a sensor |
| `get_active_alerts()` | Unacknowledged alerts, most recent first |
| `acknowledge_alert(alert_id)` | Mark an alert as handled |
| `bulk_acknowledge_alerts()` | Acknowledge multiple alerts at once; **elicits** the user for scope (all/critical-only/none) when there's more than one active alert |
| `summarize_zone_health(zone)` | Gathers a zone's sensors/alerts, then **samples** the client's LLM for a natural-language health summary |

## MCP Resources

| Resource | Description |
|----------|-------------|
| `sensors://fleet` | Live listing of all sensors and their current values |

## MCP Prompts

| Prompt | Description |
|--------|-------------|
| `diagnose_alert(alert_id)` | Structures an alert investigation: pulls the alert and its sensor's past-hour history into a template asking the model to assess cause and recommend next steps |

## Example prompts (for Claude Desktop)

- "List all sensors and their current status."
- "What's the CO2 level in Zone B right now?"
- "Show me the temperature trend for sensor 1 over the last hour."
- "Are there any active alerts? Summarize their severity."
- "Set the smoke alert threshold for sensor 6 to 0–40, then check for new alerts in a minute."

## Out of scope for v1

- Real MQTT/hardware integration — natural next step, ties into a separate edge-gateway project
- Authentication / multi-tenant access
- Web dashboard UI