From 68f562f6fdbbecb7cc64c468538ca53e4696735e Mon Sep 17 00:00:00 2001 From: Shuxin Lin Date: Mon, 6 Jul 2026 17:54:50 -0400 Subject: [PATCH 1/3] fix: define catalog csv collection Signed-off-by: Shuxin Lin --- src/couchdb/collections.json | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/couchdb/collections.json b/src/couchdb/collections.json index e99f367d..e03ab861 100644 --- a/src/couchdb/collections.json +++ b/src/couchdb/collections.json @@ -24,6 +24,23 @@ ] ] }, + "catalog": { + "format": "csv", + "indexes": [ + [ + "category" + ], + [ + "asset" + ], + [ + "failure_mode" + ], + [ + "sensor" + ] + ] + }, "iot": { "format": "json", "primary_key": [ @@ -111,4 +128,4 @@ "id_prefix": "fc", "indexes": [] } -} \ No newline at end of file +} From 47bd92e3d5fcb47cfde39932b1cccdd464518b69 Mon Sep 17 00:00:00 2001 From: Shuxin Lin Date: Mon, 6 Jul 2026 18:10:48 -0400 Subject: [PATCH 2/3] feat: add catalog utility tools Signed-off-by: Shuxin Lin --- src/servers/utilities/main.py | 152 +++++++++++++++++- src/servers/utilities/tests/test_utilities.py | 111 +++++++++++++ 2 files changed, 261 insertions(+), 2 deletions(-) diff --git a/src/servers/utilities/main.py b/src/servers/utilities/main.py index 42858783..ea6dc917 100644 --- a/src/servers/utilities/main.py +++ b/src/servers/utilities/main.py @@ -1,15 +1,19 @@ import json import logging +import os import tempfile from datetime import datetime, timezone from pathlib import Path +from typing import Any, Optional, Union from uuid import uuid4 +import couchdb3 import pendulum +from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP from pydantic import BaseModel -import os +load_dotenv() # Setup logging — default WARNING so stderr stays quiet when used as MCP server; # set LOG_LEVEL=INFO (or DEBUG) in the environment to see verbose output. @@ -19,9 +23,30 @@ logging.basicConfig(level=_log_level) logger = logging.getLogger("utilities-mcp-server") +# Configuration from environment +COUCHDB_URL = os.environ.get("COUCHDB_URL", "http://localhost:5984") +COUCHDB_USERNAME = os.environ.get("COUCHDB_USERNAME", "admin") +COUCHDB_PASSWORD = os.environ.get("COUCHDB_PASSWORD", "password") +CATALOG_DBNAME = os.environ.get("CATALOG_DBNAME", "catalog") + +try: + catalog_db = couchdb3.Database( + CATALOG_DBNAME, + url=COUCHDB_URL, + user=COUCHDB_USERNAME, + password=COUCHDB_PASSWORD, + ) + logger.info("Connected to CouchDB catalog database: %s", CATALOG_DBNAME) +except Exception as e: + logger.error("Failed to connect to CouchDB catalog database: %s", e) + catalog_db = None + mcp = FastMCP( "utilities", - instructions="General utilities: read JSON files and get current date/time.", + instructions=( + "General utilities: read JSON files, get current date/time, and query " + "asset, sensor, and failure-mode catalog data from CouchDB." + ), ) @@ -35,6 +60,18 @@ class TimeEnglishResult(BaseModel): iso: str +class ErrorResult(BaseModel): + error: str + + +class CatalogResult(BaseModel): + catalog_type: str + query: Optional[str] + total: int + entries: list[dict[str, Any]] + message: str + + # --- Helper Functions --- @@ -51,6 +88,62 @@ def get_temp_filename() -> str: return str(filepath) +def _clean_filter(value: Optional[str]) -> Optional[str]: + if value is None: + return None + value = value.strip() + return value or None + + +def _catalog_limit(limit: int) -> int: + return max(1, min(limit, 5000)) + + +def _find_catalog( + *, + catalog_type: str, + field: str, + value: Optional[str], + fields: list[str], + limit: int, + category: Optional[str] = None, +) -> Union[CatalogResult, ErrorResult]: + if catalog_db is None: + return ErrorResult(error="catalog database is not available") + + query_value = _clean_filter(value) + category_value = _clean_filter(category) + selector: dict[str, Any] = {field: query_value or {"$exists": True}} + if category_value is not None: + selector["category"] = category_value + + try: + res = catalog_db.find( + selector, + fields=fields, + limit=_catalog_limit(limit), + ) + docs = res.get("docs", []) + except Exception as e: + logger.error("Error querying %s catalog: %s", catalog_type, e) + return ErrorResult(error=str(e)) + + query_parts = [] + if query_value is not None: + query_parts.append(query_value) + if category_value is not None: + query_parts.append(f"category={category_value}") + query = ", ".join(query_parts) if query_parts else None + + return CatalogResult( + catalog_type=catalog_type, + query=query, + total=len(docs), + entries=docs, + message=f"found {len(docs)} {catalog_type} catalog entries", + ) + + # --- JSON Tools --- @@ -66,6 +159,61 @@ def json_reader(file_name: str) -> str: return json.dumps({"error": str(e)}) +# --- Catalog Tools --- + + +@mcp.tool(title="Get Sensor Catalog") +def get_sensor_catalog( + sensor: Optional[str] = None, + limit: int = 1000, +) -> Union[CatalogResult, ErrorResult]: + """Return sensor catalog entries. Pass sensor for an exact sensor-name lookup.""" + return _find_catalog( + catalog_type="sensor", + field="sensor", + value=sensor, + fields=["sensor", "description"], + limit=limit, + ) + + +@mcp.tool(title="Get Asset Catalog") +def get_asset_catalog( + asset: Optional[str] = None, + category: Optional[str] = None, + limit: int = 1000, +) -> Union[CatalogResult, ErrorResult]: + """Return asset catalog entries. Pass asset and/or category for exact lookups.""" + return _find_catalog( + catalog_type="asset", + field="asset", + value=asset, + category=category, + fields=["category", "category_description", "asset", "description"], + limit=limit, + ) + + +@mcp.tool(title="Get Failure Model Catalog") +def get_failure_model_catalog( + failure_mode: Optional[str] = None, + category: Optional[str] = None, + limit: int = 1000, +) -> Union[CatalogResult, ErrorResult]: + """Return failure-mode catalog entries. + + Pass failure_mode and/or category for exact lookups. + """ + return _find_catalog( + catalog_type="failure_mode", + field="failure_mode", + value=failure_mode, + category=category, + fields=["category", "failure_mode", "description"], + limit=limit, + ) + + # --- Time Tools --- diff --git a/src/servers/utilities/tests/test_utilities.py b/src/servers/utilities/tests/test_utilities.py index 0940ea26..2d287a05 100644 --- a/src/servers/utilities/tests/test_utilities.py +++ b/src/servers/utilities/tests/test_utilities.py @@ -5,10 +5,61 @@ import tempfile import pytest +from servers.utilities import main as utilities from servers.utilities.main import mcp from .conftest import call_tool +class FakeCatalogDB: + def __init__(self, docs): + self.docs = docs + self.calls = [] + + def find(self, selector, fields=None, limit=200): + self.calls.append({"selector": selector, "fields": fields, "limit": limit}) + matched = [] + for doc in self.docs: + if self._matches(doc, selector): + if fields: + matched.append( + {field: doc[field] for field in fields if field in doc} + ) + else: + matched.append(doc) + return {"docs": matched[:limit]} + + def _matches(self, doc, selector): + for field, expected in selector.items(): + if isinstance(expected, dict) and "$exists" in expected: + if (field in doc) != expected["$exists"]: + return False + elif doc.get(field) != expected: + return False + return True + + +@pytest.fixture +def fake_catalog_db(monkeypatch): + fake = FakeCatalogDB( + [ + {"sensor": "air flow", "description": "Airflow sensor"}, + { + "category": "rotating equipment", + "category_description": "Rotating equipment", + "asset": "Electric motor", + "description": "Converts electricity into motion.", + }, + { + "category": "rotating equipment", + "failure_mode": "Air inlet blockage", + "description": "Air intake is blocked.", + }, + ] + ) + monkeypatch.setattr(utilities, "catalog_db", fake) + return fake + + # --------------------------------------------------------------------------- # current_date_time # --------------------------------------------------------------------------- @@ -104,3 +155,63 @@ async def test_invalid_json_content(self): assert "error" in data finally: os.remove(tmp_name) + + +# --------------------------------------------------------------------------- +# catalog tools +# --------------------------------------------------------------------------- + + +class TestCatalogTools: + @pytest.mark.anyio + async def test_get_sensor_catalog_lists_sensor_entries(self, fake_catalog_db): + data = await call_tool(mcp, "get_sensor_catalog", {}) + + assert data["catalog_type"] == "sensor" + assert data["total"] == 1 + assert data["entries"] == [ + {"sensor": "air flow", "description": "Airflow sensor"} + ] + assert fake_catalog_db.calls[-1]["selector"] == { + "sensor": {"$exists": True} + } + + @pytest.mark.anyio + async def test_get_asset_catalog_filters_asset_and_category(self, fake_catalog_db): + data = await call_tool( + mcp, + "get_asset_catalog", + {"asset": "Electric motor", "category": "rotating equipment"}, + ) + + assert data["catalog_type"] == "asset" + assert data["query"] == "Electric motor, category=rotating equipment" + assert data["total"] == 1 + assert data["entries"][0]["asset"] == "Electric motor" + assert fake_catalog_db.calls[-1]["selector"] == { + "asset": "Electric motor", + "category": "rotating equipment", + } + + @pytest.mark.anyio + async def test_get_failure_model_catalog_queries_failure_mode( + self, fake_catalog_db + ): + data = await call_tool( + mcp, + "get_failure_model_catalog", + {"failure_mode": "Air inlet blockage"}, + ) + + assert data["catalog_type"] == "failure_mode" + assert data["total"] == 1 + assert data["entries"] == [ + { + "category": "rotating equipment", + "failure_mode": "Air inlet blockage", + "description": "Air intake is blocked.", + } + ] + assert fake_catalog_db.calls[-1]["selector"] == { + "failure_mode": "Air inlet blockage" + } From a5a1e635b28b4ab1fd7bcf72f23c9f63393ba95e Mon Sep 17 00:00:00 2001 From: Shuxin Lin Date: Mon, 6 Jul 2026 18:23:49 -0400 Subject: [PATCH 3/3] refactor: hide catalog query limit Signed-off-by: Shuxin Lin --- src/servers/utilities/main.py | 14 ++------------ src/servers/utilities/tests/test_utilities.py | 1 + 2 files changed, 3 insertions(+), 12 deletions(-) diff --git a/src/servers/utilities/main.py b/src/servers/utilities/main.py index ea6dc917..dd940f7c 100644 --- a/src/servers/utilities/main.py +++ b/src/servers/utilities/main.py @@ -28,6 +28,7 @@ COUCHDB_USERNAME = os.environ.get("COUCHDB_USERNAME", "admin") COUCHDB_PASSWORD = os.environ.get("COUCHDB_PASSWORD", "password") CATALOG_DBNAME = os.environ.get("CATALOG_DBNAME", "catalog") +CATALOG_QUERY_LIMIT = 1000 try: catalog_db = couchdb3.Database( @@ -95,17 +96,12 @@ def _clean_filter(value: Optional[str]) -> Optional[str]: return value or None -def _catalog_limit(limit: int) -> int: - return max(1, min(limit, 5000)) - - def _find_catalog( *, catalog_type: str, field: str, value: Optional[str], fields: list[str], - limit: int, category: Optional[str] = None, ) -> Union[CatalogResult, ErrorResult]: if catalog_db is None: @@ -121,7 +117,7 @@ def _find_catalog( res = catalog_db.find( selector, fields=fields, - limit=_catalog_limit(limit), + limit=CATALOG_QUERY_LIMIT, ) docs = res.get("docs", []) except Exception as e: @@ -165,7 +161,6 @@ def json_reader(file_name: str) -> str: @mcp.tool(title="Get Sensor Catalog") def get_sensor_catalog( sensor: Optional[str] = None, - limit: int = 1000, ) -> Union[CatalogResult, ErrorResult]: """Return sensor catalog entries. Pass sensor for an exact sensor-name lookup.""" return _find_catalog( @@ -173,7 +168,6 @@ def get_sensor_catalog( field="sensor", value=sensor, fields=["sensor", "description"], - limit=limit, ) @@ -181,7 +175,6 @@ def get_sensor_catalog( def get_asset_catalog( asset: Optional[str] = None, category: Optional[str] = None, - limit: int = 1000, ) -> Union[CatalogResult, ErrorResult]: """Return asset catalog entries. Pass asset and/or category for exact lookups.""" return _find_catalog( @@ -190,7 +183,6 @@ def get_asset_catalog( value=asset, category=category, fields=["category", "category_description", "asset", "description"], - limit=limit, ) @@ -198,7 +190,6 @@ def get_asset_catalog( def get_failure_model_catalog( failure_mode: Optional[str] = None, category: Optional[str] = None, - limit: int = 1000, ) -> Union[CatalogResult, ErrorResult]: """Return failure-mode catalog entries. @@ -210,7 +201,6 @@ def get_failure_model_catalog( value=failure_mode, category=category, fields=["category", "failure_mode", "description"], - limit=limit, ) diff --git a/src/servers/utilities/tests/test_utilities.py b/src/servers/utilities/tests/test_utilities.py index 2d287a05..c60e0ff9 100644 --- a/src/servers/utilities/tests/test_utilities.py +++ b/src/servers/utilities/tests/test_utilities.py @@ -175,6 +175,7 @@ async def test_get_sensor_catalog_lists_sensor_entries(self, fake_catalog_db): assert fake_catalog_db.calls[-1]["selector"] == { "sensor": {"$exists": True} } + assert fake_catalog_db.calls[-1]["limit"] == utilities.CATALOG_QUERY_LIMIT @pytest.mark.anyio async def test_get_asset_catalog_filters_asset_and_category(self, fake_catalog_db):