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.

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