Build a Kaiterra MCP Server for Agentic Indoor Environmental Quality Monitoring

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

On August 12, 2026, Kaiterra launched Kaiterra AI and the Kaiterra MCP for real estate, facilities, and sustainability teams. Kaiterra AI turns indoor environmental quality (IEQ) monitor readings into plain-language answers, and the Kaiterra MCP connects that data to other AI systems — so an agent can ask "which floors exceeded PM2.5 limits this week?" and get a real answer, not a dashboard URL. The launch is part of the broader shift from dashboards to workflow helpers: vendors are exposing their sensor data as agent tools rather than forcing humans to read charts. This guide builds a production-grade Python FastMCP server, kaiterra-mcp, that wraps the Kaiterra API into clean typed agent tools — live readings, trends, alerts, and portfolio health — and wires it into Claude Desktop and LangGraph. If you are building an agent tool surface around building or sensor data, the MCP directory is the reference map for the connector layer.

Why a gateway over the official MCP server

The official Kaiterra MCP is the fastest on-ramp: connect it, and an agent can query IEQ data out of the box. A custom FastMCP gateway is the right call when you need any of these:

Before the MCP launch, indoor environmental quality data lived behind vendor dashboards — useful for a facilities manager who knows where to click, invisible to every other system in the building. Kaiterra's move is part of a pattern we have tracked all year: sensor platforms are becoming agent platforms, exposing readings not as charts but as callable tools. The payoff is that the data stops being a report and starts being an input to decisions — an agent can compare a floor's CO2 trend against the HVAC schedule, flag a rising PM2.5 before it becomes an air-quality incident, and hand an operator a precise, sourced recommendation instead of a spreadsheet. That is the difference between monitoring and acting, and it is exactly the shift the broader agent workflow ecosystem is built around. For real estate portfolios with hundreds of monitors, the scale argument is even stronger: no human is watching every sensor, but an agent can watch them all, every minute, and only interrupt a human when something matters.

• A narrower, audited surface. The gateway exposes exactly the tools your facility and sustainability agents are approved to touch — reads and thresholds, not account administration.
• Combined workflows in one call. A single portfolio_health call that aggregates hundreds of monitors, or a read-then-escalate sequence, is painful to orchestrate against a raw API.
• Per-agent rate limits and keys. One key per agent or tenant means decommissioning a rogue agent means revoking a single key, not rotating a shared credential.
• Governance and audit. The gateway logs every tool call, who made it, and what it returned — the same discipline you apply to any agent workflow touching real-world data.

The tool surface and architecture

The gateway exposes five tools against the Kaiterra API:

Each tool is a thin, cached, rate-limited call to the Kaiterra API — the agent surface stays a governed view over the sensor platform, never a second source of truth.

Step 1: Scaffold the Python FastMCP server

mkdir kaiterra-mcp && cd kaiterra-mcp
python -m venv .venv && source .venv/bin/activate
pip install "fastmcp[cli]" httpx

# server.py
import os
import time
import json
import threading

import httpx
from fastmcp import FastMCP

CLIENT_ID = os.environ["KAITERRA_CLIENT_ID"]
CLIENT_SECRET = os.environ["KAITERRA_CLIENT_SECRET"]
TOKEN_URL = os.environ.get("KAITERRA_TOKEN_URL", "https://api.kaiterra.com/oauth/token")
BASE = os.environ.get("KAITERRA_BASE", "https://api.kaiterra.com/v1")
DEFAULT_TENANT = os.environ.get("KAITERRA_TENANT", "")

mcp = FastMCP("kaiterra-mcp", instructions=(
    "Indoor environmental quality tools over Kaiterra monitors. "
    "Return live values with units; prefer portfolio_health for aggregate questions."
))

_token = {"value": None, "exp": 0}
_lock = threading.Lock()

def _access_token() -> str:
    with _lock:
        if _token["value"] and _token["exp"] > time.time() + 120:
            return _token["value"]
        r = httpx.post(TOKEN_URL, data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
        }, timeout=15)
        r.raise_for_status()
        body = r.json()
        _token.update({"value": body["access_token"],
                       "exp": time.time() + body.get("expires_in", 3600)})
        return _token["value"]

def _headers():
    return {"Authorization": f"Bearer {_access_token()}", "Accept": "application/json"}

@mcp.tool()
def live_readings(location_id: str) -> str:
    """Return current IEQ readings (CO2, PM2.5, VOC, temp, humidity) for a location."""
    r = httpx.get(f"{BASE}/locations/{location_id}/readings/latest",
                  headers=_headers(), timeout=15)
    r.raise_for_status()
    return json.dumps(r.json(), indent=2)

@mcp.tool()
def get_trend(location_id: str, metric: str = "pm25",
              window: str = "day") -> str:
    """Return a time-series trend for a metric: pm25, co2, voc, temp, humidity."""
    r = httpx.get(f"{BASE}/locations/{location_id}/readings/trend",
                  params={"metric": metric, "window": window},
                  headers=_headers(), timeout=15)
    r.raise_for_status()
    return json.dumps(r.json(), indent=2)

@mcp.tool()
def list_alerts(status: str = "active", limit: int = 20) -> str:
    """List threshold alerts across the portfolio (active or recent)."""
    r = httpx.get(f"{BASE}/alerts", params={"status": status, "limit": limit},
                  headers=_headers(), timeout=15)
    r.raise_for_status()
    return json.dumps(r.json(), indent=2)

@mcp.tool()
def portfolio_health(tenant: str = "", window: str = "week") -> str:
    """Aggregated IEQ health score per floor/building/tenant over a window."""
    params = {"window": window}
    if tenant:
        params["tenant"] = tenant
    r = httpx.get(f"{BASE}/portfolio/health", params=params,
                  headers=_headers(), timeout=20)
    r.raise_for_status()
    return json.dumps(r.json(), indent=2)

@mcp.tool()
def get_location(location_id: str) -> str:
    """Return monitor and location metadata for a location."""
    r = httpx.get(f"{BASE}/locations/{location_id}",
                  headers=_headers(), timeout=15)
    r.raise_for_status()
    return json.dumps(r.json(), indent=2)

if __name__ == "__main__":
    mcp.run()

Run the server with python server.py and it speaks MCP over stdio to whichever client you wire next.

Step 2: inputSchema definitions published to agents

FastMCP derives JSON Schema from Python type hints, but pinning the contract helps agents call tools correctly:

{
  "live_readings": {
    "type": "object",
    "properties": {
      "location_id": { "type": "string", "description": "Kaiterra location or monitor ID" }
    },
    "required": ["location_id"]
  },
  "get_trend": {
    "type": "object",
    "properties": {
      "location_id": { "type": "string" },
      "metric": { "type": "string", "enum": ["pm25", "co2", "voc", "temp", "humidity"], "default": "pm25" },
      "window": { "type": "string", "enum": ["hour", "day", "week"], "default": "day" }
    },
    "required": ["location_id"]
  },
  "portfolio_health": {
    "type": "object",
    "properties": {
      "tenant": { "type": "string", "default": "", "description": "Tenant or workspace scope; empty = all" },
      "window": { "type": "string", "enum": ["day", "week", "month"], "default": "week" }
    }
  }
}

Descriptions matter ten times more in agent-facing schemas than in human API docs — the model chooses a tool off the description alone, so say what each tool returns and the units in every description string.

Step 3: Wire into Claude Desktop and Cursor

{
  "mcpServers": {
    "kaiterra-mcp": {
      "command": "python",
      "args": ["/absolute/path/to/kaiterra-mcp/server.py"],
      "env": {
        "KAITERRA_CLIENT_ID": "your-client-id",
        "KAITERRA_CLIENT_SECRET": "your-client-secret",
        "KAITERRA_TENANT": "acme-portfolio"
      }
    }
  }
}

The client secret belongs in your secret manager on the machine running the gateway, never in a committed config. From the Claude Code CLI you can add it with claude mcp add kaiterra -- python /absolute/path/to/kaiterra-mcp/server.py and then confirm with "What MCP tools do you have available?".

OAuth 2.0, rate limits, and caching

• Client-credentials flow. The gateway authenticates as a registered application, caching tokens until near expiry. Rotate the client secret centrally and audit every application that has one.
• Per-tenant scoping. Issue one client ID per tenant or workspace so a portfolio-level agent and a single-building agent see different data worlds. The KAITERRA_TENANT env var pins the default scope.
• Token-bucket rate limiting. Cap each agent at a modest request rate so a runaway polling loop cannot hammer the API or burn your data quota. Add a per-agent limiter like the one in our MCP server builds.
• TTL caching. Live readings change every minute, but a 30-second cache on live_readings and a 5-minute cache on get_trend cut API calls dramatically for always-on agents. Short TTLs keep freshness while absorbing burst patterns.
• Prompt-injection guard. Treat location IDs and metric names as untrusted input; validate them against the location registry before hitting the API. Pair the gateway with the same screening you apply to your other production agent surfaces.

Step 4: Wire into a LangGraph facilities agent

The gateway is a plain MCP server, so orchestration frameworks connect with the standard adapters:

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import ToolNode
from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
from contextlib import AsyncExitStack

async def tool_node():
    stack = AsyncExitStack()
    params = StdioServerParameters(
        command="python",
        args=["/absolute/path/to/kaiterra-mcp/server.py"],
        env={**os.environ})
    read, write = await stack.enter_async_context(stdio_client(params))
    tools = await load_mcp_tools(read, write)
    return ToolNode(tools)

Inside a LangGraph loop, a facilities agent can watch portfolio_health, call list_alerts, drill into get_trend for a rising CO2 floor, and escalate to a human operator with the exact readings attached — the same read-then-act pattern you see across the AI workflows library. That is where Kaiterra's launch stops being a data connector and becomes a workflow helper: the agent does the monitoring and the escalation, and the human does the decision.

Testing the server end to end

python -m fastmcp inspect "$(pwd)/server.py"

> live_readings("floor-3-east")
  → co2: 812 ppm, pm25: 9.4 µg/m³, voc: 0.31 mg/m³, temp: 23.1°C, humidity: 41%

> list_alerts("active")
  → 2 alerts: floor-3-east PM2.5 trending up, lobby CO2 above 1000 ppm

> portfolio_health("acme-portfolio", "week")
  → 92 / 100 portfolio health; 3 locations below 80

Then repeat the same prompt in Claude Desktop with the server attached: the second live_readings call should return from cache in single-digit milliseconds — proof the gateway, not the host, is doing the work.

Frequently Asked Questions

What did Kaiterra announce on August 12, 2026?

Kaiterra launched Kaiterra AI and the Kaiterra MCP on August 12, 2026. Kaiterra AI answers plain-language questions about indoor environmental quality from a customer's own monitor readings, and the Kaiterra MCP connects that data to other AI systems.

Which environmental metrics does a Kaiterra MCP server expose?

The core monitored metrics are CO2, PM2.5, PM10, VOC, temperature, humidity, and pressure — the standard IEQ dashboard set, exposed as typed agent tools with live values, trends, and thresholds.

How do I secure a Kaiterra MCP server for facility agents?

Use OAuth 2.0 client-credentials flow with per-tenant client IDs, scope API tokens to specific workspaces or locations, apply a token-bucket rate limiter per agent, cache readings with a short TTL, and keep credentials in a secret manager.

Should I use the official Kaiterra MCP or build my own gateway?

Use the official Kaiterra MCP for a quick start. Build a custom FastMCP gateway when you need a narrower approved tool surface, combined read-then-act workflows in a single call, per-agent rate limits and keys, or integration with your existing LangGraph agent stack.

What is a realistic agent use case for Kaiterra data?

A facilities agent that watches CO2 and PM2.5 across a portfolio, detects readings above threshold, correlates them with HVAC schedules, and escalates to an operator — or a sustainability agent that reports portfolio-wide IEQ health for compliance and wellness programs.

Closing thoughts

Kaiterra's August launch is the clearest sign yet that sensor platforms are becoming agent platforms: the data is no longer trapped in a dashboard, it is a tool an agent can call. Production use still demands the boring engineering around the API — vaulted credentials, per-tenant scopes, rate limits, and caches — but the payoff is an agent that watches your buildings the way a good facilities manager would. Track more builds like this one in the MCP directory and keep an eye on AI news as more building and sensor platforms follow Kaiterra's lead.