Skip to main content
Glama
csaad-ai-engineer

FastMCP IoT Sensor Server

FastMCP IoT Sensor Server

A portfolio artifact demonstrating hands-on MCP (Model Context Protocol) server development using FastMCP. 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; for a deeper file-by-file walkthrough (useful for talking through this project in an interview) see docs/project-walkthrough.md; for the MCP protocol/design background behind the decisions here see 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 below), not part of the server's own runtime.

Related MCP server: MCP Weather Server — Demo

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.

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:

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

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

    pytest tests/ -v
  2. Start the server in one terminal:

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

    python -m iot_sensors.simulation.run
  4. Interact with it via the Inspector (instead of step 2 — the Inspector launches the server for you), Claude Desktop (see below), or any other MCP client.

Run the server standalone (stdio transport)

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

or, without the config file:

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

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

python -m iot_sensors.simulation.run

Or opt the server's own lifespan into starting it:

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

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

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

Related MCP Connectors

Related MCP Servers

  • F
    license
    Not graded
    quality
    D
    maintenance
    MCP server that bridges MQTT sensor data to MCP, allowing clients to query cached sensor readings and send commands via tools like list_topics, read_topic, and publish.
    -
  • A
    license
    A
    quality
    B
    maintenance
    Exposes live industrial IoT telemetry to any MCP client, streaming simulated sensor data from a fleet of machines and detecting anomalies, with the ability to inject faults on demand.
    4
    MIT