base implementation

Author: daimor

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.venv

.python-version

@@ -0,0 +1 @@
+3.12

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2021 Dmitry Maslennikov
+Copyright (c) 2025 Dmitry Maslennikov
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

pyproject.toml

@@ -0,0 +1,39 @@
+[project]
+name = "mcp-server-iris"
+version = "0.1.0"
+description = "A Model Context Protocol server for InterSystems IRIS."
+readme = "README.md"
+requires-python = ">=3.10"
+authors = [{ name = "CaretDev Corp." }]
+maintainers = [{ name = "Dmitry Maslennikov", email = "[email protected]" }]
+keywords = ["iris", "mcp", "llm", "automation"]
+license = { text = "MIT" }
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: MIT License",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.10",
+]
+dependencies = [
+    "intersystems-irispython>=5.1.0",
+    "mcp[cli]>=1.2.0",
+    "starlette>=0.36.0",
+    "uvicorn>=0.27.0",
+]
+
+[project.scripts]
+mcp-server-iris = "mcp_server_iris:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.uv]
+dev-dependencies = ["pyright>=1.1.389", "ruff>=0.7.3", "pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+python_files = "test_*.py"
+python_classes = "Test*"
+python_functions = "test_*"

src/mcp_server_iris/__init__.py

@@ -0,0 +1,2 @@
+from .server import main as main # noqa
+# from .server import run as run # noqa

src/mcp_server_iris/__main__.py

@@ -0,0 +1,3 @@
+from mcp_server_iris import main
+
+main()

src/mcp_server_iris/interoperability.py

@@ -0,0 +1,214 @@
+from enum import Enum
+from mcp_server_iris.mcpserver import Context
+from mcp_server_iris.tools import transaction, raise_on_error
+from iris import IRISReference
+
+
+class ProductionStatus(Enum):
+    Unknown = 0
+    Running = 1
+    Stopped = 2
+    Suspended = 3
+    Troubled = 4
+    NetworkStopped = 5
+    ShardWorkerProhibited = 6
+
+
+log_types = [
+    "ASSERT",
+    "ERROR",
+    "WARNING",
+    "INFO",
+    "TRACE",
+    "ALERT",
+]
+
+
+class LogType(Enum):
+    Assert = 1
+    Error = 2
+    Warning = 3
+    Info = 4
+    Trace = 5
+    Alert = 6
+
+
+def production_items_status(iris, running: bool, name: str) -> list[str]:
+    result = []
+    namespace = iris.classMethodString("%SYSTEM.Process", "NameSpace")
+    prod = iris.classMethodObject("Ens.Config.Production", "%OpenId", name)
+    if not prod:
+        raise ValueError(f"Production {name} not found")
+    items = prod.getObject("Items")
+    for i in range(1, items.invokeInteger("Count") + 1):
+        item = items.invokeObject("GetAt", i)
+        item_name = item.getString("Name")
+        status_info = []
+        enabled = item.getBoolean("Enabled")
+        status_info += [f"Enabled={enabled}"]
+        if enabled:
+            val = iris.getString(
+                "^IRIS.Temp.EnsHostMonitor", namespace, item_name, "%Status"
+            )
+            status_info += [f"Status={val}"]
+
+        result.append(f"{item_name}: " + "; ".join(status_info))
+    return result
+
+
+def init(server, logger):
+    @server.tool(description="Create an Interoperability Production")
+    async def interoperability_production_create(name: str, ctx: Context) -> bool:
+        if "." not in name:
+            raise ValueError(
+                "Production name must in format packagenamespace.productionname, where packagenamespace can have multiple parts separated by dots"
+            )
+        iris = ctx.request_context.lifespan_context["iris"]
+        with transaction(iris):
+            prod = iris.classMethodObject(
+                "%Dictionary.ClassDefinition", "%OpenId", name
+            )
+            if prod:
+                raise ValueError(f"Class {name} already exists")
+            logger.info(f"Creating Interoperability Production: {name}")
+            prod = iris.classMethodObject("Ens.Config.Production", "%New", name)
+            raise_on_error(iris, prod.invokeString("SaveToClass"))
+            raise_on_error(iris, prod.invokeString("%Save"))
+            raise_on_error(
+                iris, iris.classMethodString("%SYSTEM.OBJ", "Compile", name, "ck-d")
+            )
+            return True
+
+    @server.tool(description="Status of an Interoperability Production")
+    async def interoperability_production_status(
+        ctx: Context,
+        name: str = None,
+        full_status: bool = False,
+    ) -> str:
+        logger.info("Interoperability Production Status" + f": {name}" if name else "")
+        iris = ctx.request_context.lifespan_context["iris"]
+        refname = IRISReference(iris)
+        refname.setValue(name)
+        refstatus = IRISReference(iris)
+        raise_on_error(
+            iris,
+            iris.classMethodString(
+                "Ens.Director", "GetProductionStatus", refname, refstatus
+            ),
+        )
+        if not refname.getValue():
+            raise ValueError("No running production found")
+        name = refname.getValue()
+        status = ProductionStatus(int(refstatus.getValue()))
+        reason = IRISReference(iris)
+        needsupdate = iris.classMethodBoolean(
+            "Ens.Director", "ProductionNeedsUpdate", reason
+        )
+        reason_update = (
+            f"Production needs update: {reason.getValue()}" if needsupdate else ""
+        )
+
+        if status == ProductionStatus.Running and full_status:
+            items_status = production_items_status(
+                iris, status == ProductionStatus.Running, name
+            )
+            return f"Production {name} is running with items: \n{"\n".join(items_status)}\n{reason_update}"
+        return f"Production {name} with status: {status.name}\n{reason_update}"
+
+    @server.tool(description="Start an Interoperability Production")
+    async def interoperability_production_start(
+        ctx: Context,
+        name: str = None,
+    ) -> str:
+        logger.info(
+            "Starting Interoperability Production" + f": {name}" if name else "."
+        )
+        iris = ctx.request_context.lifespan_context["iris"]
+        raise_on_error(
+            iris, iris.classMethodString("Ens.Director", "StartProduction", name)
+        )
+        return "Started production"
+
+    @server.tool(description="Stop an Interoperability Production")
+    async def interoperability_production_stop(
+        ctx: Context,
+        timeout: int = None,
+        force: bool = False,
+    ) -> str:
+        logger.info("Sopping Interoperability Production.")
+        iris = ctx.request_context.lifespan_context["iris"]
+        raise_on_error(
+            iris,
+            iris.classMethodString("Ens.Director", "StopProduction", timeout, force),
+        )
+        return "Stopped production"
+
+    @server.tool(description="Recover an Interoperability Production")
+    async def interoperability_production_recover(
+        ctx: Context,
+    ) -> str:
+        logger.info("Recovering Interoperability Production")
+        iris = ctx.request_context.lifespan_context["iris"]
+        raise_on_error(
+            iris, iris.classMethodString("Ens.Director", "RecoverProduction")
+        )
+        return "Recovered"
+
+    @server.tool(description="Check if an Interoperability Production needs update")
+    async def interoperability_production_needsupdate(
+        ctx: Context,
+    ) -> str:
+        logger.info("Checking if Interoperability Production needs update")
+        iris = ctx.request_context.lifespan_context["iris"]
+        reason = IRISReference(iris)
+        result = iris.classMethodBoolean(
+            "Ens.Director", "ProductionNeedsUpdate", reason
+        )
+        if result:
+            raise ValueError(f"Production needs update: {reason.getValue()}")
+        return "Production does not need update"
+
+    @server.tool(description="Update Interoperability Production")
+    async def interoperability_production_update(
+        ctx: Context,
+        timeout: int = None,
+        force: bool = False,
+    ) -> str:
+        iris = ctx.request_context.lifespan_context["iris"]
+        raise_on_error(
+            iris,
+            iris.classMethodString("Ens.Director", "UpdateProduction", timeout, force),
+        )
+        return "Production updated"
+
+    @server.tool(description="Get Interoperability Production logs")
+    async def interoperability_production_logs(
+        ctx: Context,
+        item_name: str = None,
+        limit: int = 10,
+        log_type_info: bool = False,
+        log_type_alert: bool = False,
+        log_type_error: bool = True,
+        log_type_warning: bool = True,
+    ) -> str:
+        logs = []
+        log_type = []
+        log_type_info and log_type.append(LogType.Info.value)
+        log_type_alert and log_type.append(LogType.Alert.value)
+        log_type_error and log_type.append(LogType.Error.value)
+        log_type_warning and log_type.append(LogType.Warning.value)
+        db = ctx.request_context.lifespan_context["db"]
+        with db.cursor() as cur:
+            sql = f"""
+select top ? TimeLogged , %External(Type) Type, ConfigName, Text
+from Ens_Util.Log
+where
+{"ConfigName = ?" if item_name else "1=1"}
+{f"and type in ({', '.join(['?'] * len(log_type))})" if log_type else ""}
+order by id desc
+"""
+            params = [limit, *([item_name] if item_name else []), *log_type]
+            cur.execute(sql, params)
+            for row in cur.fetchall():
+                logs.append(f"{row[0]} {row[1]} {row[2]} {row[3]}")
+        return "\n".join(logs)