|
| 1 | +## |
| 2 | +## © Copyright 2021- IBM Inc. All rights reserved |
| 3 | +# SPDX-License-Identifier: MIT |
| 4 | +## |
| 5 | + |
| 6 | +""" |
| 7 | +MCP (Model Context Protocol) server for elmclient. |
| 8 | +
|
| 9 | +Exposes ELM functionality as tools for AI assistants. |
| 10 | +See https://modelcontextprotocol.io/ for the MCP specification. |
| 11 | +
|
| 12 | +Usage: |
| 13 | + elm-mcp-server # stdio transport (default) |
| 14 | + elm-mcp-server --transport sse # SSE transport |
| 15 | +
|
| 16 | +Configuration via environment variables: |
| 17 | + ELM_HOST - Jazz server URL (required) |
| 18 | + ELM_USER - Username (required) |
| 19 | + ELM_PASSWORD - Password (required) |
| 20 | + ELM_JTS_CONTEXT - JTS context path (default: 'jts') |
| 21 | + ELM_VERIFY_SSL - Verify SSL certificates (default: 'true') |
| 22 | +
|
| 23 | +Alternatively, provide a JSON credentials file at ~/.elm_credentials.json: |
| 24 | + {"host": "https://...", "username": "...", "password": "..."} |
| 25 | +""" |
| 26 | + |
| 27 | +import json |
| 28 | +import logging |
| 29 | +import os |
| 30 | +import sys |
| 31 | + |
| 32 | +try: |
| 33 | + from mcp.server.fastmcp import FastMCP |
| 34 | +except ImportError: |
| 35 | + print( |
| 36 | + "MCP server requires the 'mcp' package. " |
| 37 | + "Install with: pip install 'elmclient[mcp]'", |
| 38 | + file=sys.stderr, |
| 39 | + ) |
| 40 | + sys.exit(1) |
| 41 | + |
| 42 | +import elmclient.server as elmserver |
| 43 | +import elmclient.rdfxml as rdfxml |
| 44 | + |
| 45 | +logger = logging.getLogger(__name__) |
| 46 | + |
| 47 | +mcp = FastMCP( |
| 48 | + "elm", |
| 49 | + instructions=( |
| 50 | + "IBM ELM (Engineering Lifecycle Management) server. " |
| 51 | + "Query work items, requirements, and test artifacts." |
| 52 | + ), |
| 53 | +) |
| 54 | + |
| 55 | +OSLC_PREFIXES = { |
| 56 | + "http://purl.org/dc/terms/": "dcterms", |
| 57 | + "http://open-services.net/ns/cm#": "oslc_cm", |
| 58 | +} |
| 59 | + |
| 60 | +CRED_FILE = os.path.expanduser("~/.elm_credentials.json") |
| 61 | + |
| 62 | +_connections: dict = {} |
| 63 | + |
| 64 | + |
| 65 | +def _load_config() -> dict: |
| 66 | + """Load connection config from environment or credentials file.""" |
| 67 | + host = os.environ.get("ELM_HOST") |
| 68 | + if host: |
| 69 | + return { |
| 70 | + "host": host, |
| 71 | + "username": os.environ["ELM_USER"], |
| 72 | + "password": os.environ["ELM_PASSWORD"], |
| 73 | + "jts_context": os.environ.get("ELM_JTS_CONTEXT", "jts"), |
| 74 | + "verify_ssl": os.environ.get("ELM_VERIFY_SSL", "true").lower() == "true", |
| 75 | + } |
| 76 | + if os.path.exists(CRED_FILE): |
| 77 | + with open(CRED_FILE) as f: |
| 78 | + data = json.load(f) |
| 79 | + return { |
| 80 | + "host": data["host"], |
| 81 | + "username": data["username"], |
| 82 | + "password": data["password"], |
| 83 | + "jts_context": data.get("jts_context", "jts"), |
| 84 | + "verify_ssl": data.get("verify_ssl", True), |
| 85 | + } |
| 86 | + raise RuntimeError( |
| 87 | + f"ELM credentials not found. Set ELM_HOST/ELM_USER/ELM_PASSWORD " |
| 88 | + f"environment variables or create {CRED_FILE}" |
| 89 | + ) |
| 90 | + |
| 91 | + |
| 92 | +def _get_server(): |
| 93 | + if "server" not in _connections: |
| 94 | + cfg = _load_config() |
| 95 | + _connections["server"] = elmserver.JazzTeamServer( |
| 96 | + cfg["host"], |
| 97 | + cfg["username"], |
| 98 | + cfg["password"], |
| 99 | + verifysslcerts=cfg["verify_ssl"], |
| 100 | + jtsappstring=cfg["jts_context"], |
| 101 | + appstring="ccm", |
| 102 | + cachingcontrol=0, |
| 103 | + ) |
| 104 | + _connections["config"] = cfg |
| 105 | + return _connections["server"] |
| 106 | + |
| 107 | + |
| 108 | +def _get_app(domain: str = "ccm"): |
| 109 | + key = f"app_{domain}" |
| 110 | + if key not in _connections: |
| 111 | + server = _get_server() |
| 112 | + _connections[key] = server.find_app(domain, ok_to_create=True) |
| 113 | + return _connections[key] |
| 114 | + |
| 115 | + |
| 116 | +def _find_query_base(project): |
| 117 | + """Find the OSLC query base URI for a project.""" |
| 118 | + services_xml = project.get_services_xml() |
| 119 | + qcaps = rdfxml.xml_find_elements( |
| 120 | + services_xml, |
| 121 | + ".//{http://open-services.net/ns/core#}QueryCapability", |
| 122 | + ) |
| 123 | + for qc in qcaps: |
| 124 | + qb = rdfxml.xmlrdf_get_resource_uri(qc, "oslc:queryBase") |
| 125 | + if qb: |
| 126 | + return qb |
| 127 | + return None |
| 128 | + |
| 129 | + |
| 130 | +@mcp.tool() |
| 131 | +def list_projects(domain: str = "ccm") -> str: |
| 132 | + """List accessible projects in ELM. |
| 133 | +
|
| 134 | + Args: |
| 135 | + domain: 'ccm' (EWM/work items), 'rm' (DOORS Next/requirements), 'qm' (ETM/tests) |
| 136 | + """ |
| 137 | + app = _get_app(domain) |
| 138 | + app._load_projects() |
| 139 | + projects = [] |
| 140 | + for uri, info in app._projects.items(): |
| 141 | + if isinstance(info, dict): |
| 142 | + projects.append({"name": info.get("name", "?"), "uri": uri}) |
| 143 | + return json.dumps(projects, ensure_ascii=False, indent=2) |
| 144 | + |
| 145 | + |
| 146 | +@mcp.tool() |
| 147 | +def list_workitems( |
| 148 | + project_name: str, |
| 149 | + pagesize: int = 30, |
| 150 | + query: str = "", |
| 151 | +) -> str: |
| 152 | + """List work items from an EWM project. |
| 153 | +
|
| 154 | + Args: |
| 155 | + project_name: Name of the EWM project |
| 156 | + pagesize: Number of results (default 30) |
| 157 | + query: Optional OSLC query filter |
| 158 | + """ |
| 159 | + app = _get_app("ccm") |
| 160 | + project = app.find_project(project_name) |
| 161 | + if not project: |
| 162 | + return json.dumps({"error": f"Project '{project_name}' not found"}) |
| 163 | + |
| 164 | + query_base = _find_query_base(project) |
| 165 | + if not query_base: |
| 166 | + return json.dumps({"error": "Query capability not found"}) |
| 167 | + |
| 168 | + kwargs = dict( |
| 169 | + select=["dcterms:identifier", "dcterms:title"], |
| 170 | + orderbys=["-dcterms:modified"], |
| 171 | + prefixes=OSLC_PREFIXES, |
| 172 | + pagesize=pagesize, |
| 173 | + ) |
| 174 | + if query: |
| 175 | + kwargs["whereterms"] = query |
| 176 | + |
| 177 | + results = project.execute_oslc_query(query_base, **kwargs) |
| 178 | + |
| 179 | + items = [] |
| 180 | + for uri, attrs in results.items(): |
| 181 | + items.append({ |
| 182 | + "id": attrs.get("dcterms:identifier", "?"), |
| 183 | + "title": attrs.get("dcterms:title", "?"), |
| 184 | + "uri": uri, |
| 185 | + }) |
| 186 | + return json.dumps(items, ensure_ascii=False, indent=2) |
| 187 | + |
| 188 | + |
| 189 | +@mcp.tool() |
| 190 | +def get_workitem( |
| 191 | + project_name: str, |
| 192 | + workitem_id: str, |
| 193 | +) -> str: |
| 194 | + """Get details of a specific work item by ID. |
| 195 | +
|
| 196 | + Args: |
| 197 | + project_name: Name of the EWM project |
| 198 | + workitem_id: Work item identifier (e.g. '12345') |
| 199 | + """ |
| 200 | + app = _get_app("ccm") |
| 201 | + project = app.find_project(project_name) |
| 202 | + if not project: |
| 203 | + return json.dumps({"error": f"Project '{project_name}' not found"}) |
| 204 | + |
| 205 | + query_base = _find_query_base(project) |
| 206 | + if not query_base: |
| 207 | + return json.dumps({"error": "Query capability not found"}) |
| 208 | + |
| 209 | + results = project.execute_oslc_query( |
| 210 | + query_base, |
| 211 | + whereterms=[["dcterms:identifier", "=", f'"{workitem_id}"']], |
| 212 | + select=["*"], |
| 213 | + prefixes=OSLC_PREFIXES, |
| 214 | + pagesize=1, |
| 215 | + ) |
| 216 | + |
| 217 | + if not results: |
| 218 | + return json.dumps({"error": f"Work item {workitem_id} not found"}) |
| 219 | + |
| 220 | + uri, attrs = next(iter(results.items())) |
| 221 | + clean = {"uri": uri} |
| 222 | + for k, v in attrs.items(): |
| 223 | + clean[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v |
| 224 | + return json.dumps(clean, ensure_ascii=False, indent=2) |
| 225 | + |
| 226 | + |
| 227 | +def main(): |
| 228 | + mcp.run() |
| 229 | + |
| 230 | + |
| 231 | +if __name__ == "__main__": |
| 232 | + main() |
0 commit comments