Template best practices

.env.template

@@ -0,0 +1,39 @@
+# Assisted Service MCP Server - .env.template
+# Copy to `.env` and adjust values as needed. All variables are optional unless noted.
+
+# --- Server ---
+# Bind host for the HTTP/SSE server
+MCP_HOST=0.0.0.0
+# Port for the HTTP/SSE server (1024-65535)
+MCP_PORT=8000
+# Transport protocol for MCP server: sse | streamable-http
+TRANSPORT=sse
+
+# --- Assisted Service API ---
+# Assisted Service API base URL
+INVENTORY_URL=https://api.openshift.com/api/assisted-install/v2
+# Enable verbose client logging (true/false)
+CLIENT_DEBUG=false
+# URL endpoint used to fetch the pull secret
+PULL_SECRET_URL=https://api.openshift.com/api/accounts_mgmt/v1/access_token
+
+# --- Authentication ---
+# OCM offline token used to mint access tokens.
+# You may omit this and instead provide the token per-request via the
+# "OCM-Offline-Token" HTTP header from your MCP client configuration.
+# OFFLINE_TOKEN=
+
+# Red Hat SSO token endpoint used to exchange the offline token
+SSO_URL=https://sso.redhat.com/auth/realms/redhat-external/protocol/openid-connect/token
+
+# --- Logging ---
+# Logging level: DEBUG | INFO | WARNING | ERROR | CRITICAL
+LOGGING_LEVEL=INFO
+# Logger name (leave empty for default)
+# LOGGER_NAME=assisted-service-mcp
+# Write logs to file (true/false). Consider disabling in containers.
+LOG_TO_FILE=true
+
+# --- Features ---
+# Enable troubleshooting tool calls: 0 (disabled) | 1 (enabled)
+ENABLE_TROUBLESHOOTING_TOOLS=0

Dockerfile

@@ -11,11 +11,7 @@ COPY pyproject.toml .
 COPY uv.lock .
 RUN uv sync
 
-COPY server.py .
-COPY service_client ./service_client/
-COPY static_net ./static_net/
-COPY log_analyzer ./log_analyzer/
-COPY metrics ./metrics/
+COPY assisted_service_mcp ./assisted_service_mcp/
 
 RUN chown -R 1001:0 ${APP_HOME}
 
@@ -26,4 +22,4 @@ ENV LOG_TO_FILE=false
 
 EXPOSE 8000
 
-CMD ["uv", "--cache-dir", "/tmp/uv-cache", "run", "server.py"]
+CMD ["uv", "--cache-dir", "/tmp/uv-cache", "run", "python", "-m", "assisted_service_mcp.src.main"]

Makefile

@@ -15,7 +15,7 @@ run:
 
 .PHONY: run-local
 run-local:
-	uv run server.py
+	uv run python -m assisted_service_mcp.src.main
 
 .PHONY: run-mock-assisted
 run-mock-assisted:
@@ -30,7 +30,7 @@ deploy-template:
 	scripts/deploy_from_template.sh
 
 test-coverage:
-	uv run --group test pytest --cov=service_client --cov=server --cov-report=html --cov-report=term-missing
+	uv run --group test pytest --cov=assisted_service_mcp --cov-report=html --cov-report=term-missing
 
 test-verbose:
 	uv run --group test pytest -v

README.md

@@ -29,7 +29,7 @@ In VSCode for example:
                     "run",
                     "mcp",
                     "run",
-                    "/path/to/assisted-service-mcp/server.py"
+                    "/path/to/assisted-service-mcp/assisted_service_mcp/src/main.py"
                 ],
                 "env": {
                     "OFFLINE_TOKEN": <your token>
@@ -43,7 +43,7 @@ For SSE (recommended):
 
 Start the server in a terminal:
 
-`OFFLINE_TOKEN=<your token> uv run server.py`
+`OFFLINE_TOKEN=<your token> uv run assisted_service_mcp.src.main`
 
 Configure the server in the client:
 

assisted_service_mcp/__init__.py

@@ -0,0 +1,6 @@
+"""Assisted Service MCP Server.
+
+MCP server for interacting with the OpenShift assisted installer API.
+"""
+
+__version__ = "0.1.0"

assisted_service_mcp/src/__init__.py

@@ -0,0 +1 @@
+"""Source code for Assisted Service MCP Server."""

assisted_service_mcp/src/api.py

@@ -0,0 +1,23 @@
+"""FastAPI application setup for the Assisted Service MCP server.
+
+This module initializes the FastAPI app and sets up the MCP server
+with appropriate transport protocols.
+"""
+
+from assisted_service_mcp.src.mcp import AssistedServiceMCPServer
+from assisted_service_mcp.src.settings import settings
+from assisted_service_mcp.src.logger import log, configure_logging
+
+# Ensure logging is configured before any module-level log usage
+configure_logging()
+
+# Initialize the MCP server
+server = AssistedServiceMCPServer()
+
+# Choose the appropriate transport protocol based on settings
+if settings.TRANSPORT == "streamable-http":
+    app = server.mcp.streamable_http_app()
+    log.info("Using StreamableHTTP transport (stateless)")
+else:
+    app = server.mcp.sse_app()
+    log.info("Using SSE transport (stateful)")

assisted_service_mcp/src/logger.py

@@ -0,0 +1,178 @@
+"""
+Logging utilities with sensitive information filtering.
+
+This module provides logging configuration and formatting utilities that
+automatically filter sensitive information like pull secrets, SSH keys,
+and vSphere credentials from log messages.
+"""
+
+# -*- coding: utf-8 -*-
+import logging
+import re
+import sys
+
+
+class SensitiveFormatter(logging.Formatter):
+    """Formatter that removes sensitive info."""
+
+    # Default log format used by this formatter
+    DEFAULT_FORMAT = "%(asctime)s - %(name)s - %(levelname)-8s - %(thread)d:%(process)d - %(message)s - (%(pathname)s:%(lineno)d)->%(funcName)s"
+
+    def __init__(self, fmt: str | None = None) -> None:
+        """Initialize with default format if none provided."""
+        if fmt is None:
+            fmt = self.DEFAULT_FORMAT
+        super().__init__(fmt)
+
+    @staticmethod
+    def _filter(s: str) -> str:
+        # Dict filter
+        s = re.sub(r"('_pull_secret':\s+)'(.*?)'", r"\g<1>'*** PULL_SECRET ***'", s)
+        s = re.sub(r"('_ssh_public_key':\s+)'(.*?)'", r"\g<1>'*** SSH_KEY ***'", s)
+        s = re.sub(
+            r"('_vsphere_username':\s+)'(.*?)'", r"\g<1>'*** VSPHERE USER ***'", s
+        )
+        s = re.sub(
+            r"('_vsphere_password':\s+)'(.*?)'", r"\g<1>'*** VSPHERE PASSWORD ***'", s
+        )
+
+        # Object filter
+        def _redact_value(text: str, key: str, placeholder: str) -> str:
+            # Match quoted (single or double) or unquoted values, preserving spacing and quotes
+            pattern = re.compile(
+                rf"({re.escape(key)})(\s*=\s*)(?:'([^']*)'|\"([^\"]*)\"|([^\s,}}]+))"
+            )
+
+            def _repl(m: re.Match) -> str:
+                key_part = m.group(1)
+                eq_spaces = m.group(2)
+                if m.group(3) is not None:  # single-quoted
+                    return f"{key_part}{eq_spaces}'{placeholder}'"
+                if m.group(4) is not None:  # double-quoted
+                    return f'{key_part}{eq_spaces}"{placeholder}"'
+                # unquoted
+                return f"{key_part}{eq_spaces}{placeholder}"
+
+            return pattern.sub(_repl, text)
+
+        s = _redact_value(s, "pull_secret", "*** PULL_SECRET ***")
+        s = _redact_value(s, "ssh_public_key", "*** SSH_KEY ***")
+        s = _redact_value(s, "vsphere_username", "*** VSPHERE_USER ***")
+        s = _redact_value(s, "vsphere_password", "*** VSPHERE_PASSWORD ***")
+
+        return s
+
+    def format(self, record: logging.LogRecord) -> str:
+        """
+        Format log record while filtering sensitive information.
+
+        Args:
+            record: The LogRecord instance to be formatted.
+
+        Returns:
+            str: The formatted log message with sensitive info redacted.
+        """
+        original = logging.Formatter.format(self, record)
+        return self._filter(original)
+
+
+def get_logging_level() -> int:
+    """
+    Get the logging level from settings.
+
+    Returns:
+        int: The logging level (defaults to INFO if not set or invalid).
+    """
+    # Import here to avoid circular dependency at module load time
+    from assisted_service_mcp.src.settings import settings
+
+    level = settings.LOGGING_LEVEL
+    return getattr(logging, str(level).upper(), logging.INFO) if level else logging.INFO
+
+
+logging.getLogger("requests").setLevel(logging.ERROR)
+logging.getLogger("urllib3").setLevel(logging.ERROR)
+logging.getLogger("asyncio").setLevel(logging.ERROR)
+
+
+def add_log_file_handler(logger: logging.Logger, filename: str) -> logging.FileHandler:
+    """
+    Add a file handler to the logger with sensitive information filtering.
+
+    Args:
+        logger: The logger instance to add the handler to.
+        filename: The path to the log file.
+
+    Returns:
+        logging.FileHandler: The created file handler.
+    """
+    fh = logging.FileHandler(filename)
+    fh.setFormatter(SensitiveFormatter())
+    logger.addHandler(fh)
+    return fh
+
+
+def add_stream_handler(logger: logging.Logger) -> None:
+    """
+    Add a stream handler to the logger with sensitive information filtering.
+
+    Args:
+        logger: The logger instance to add the handler to.
+    """
+    ch = logging.StreamHandler(sys.stderr)
+    ch.setFormatter(SensitiveFormatter())
+    logger.addHandler(ch)
+
+
+# Export a module-level logger with a safe default name to avoid circular imports.
+# Configuration should be done by calling configure_logging() after settings are ready.
+log = logging.getLogger("assisted-service-mcp")
+
+
+def configure_logging() -> logging.Logger:
+    """
+    Configure logging after settings are available.
+
+    This sets logger names/levels, third-party logger levels, and attaches
+    file/stream handlers with the SensitiveFormatter. Importing settings here
+    avoids circular imports at module load time.
+
+    Returns:
+        logging.Logger: The configured application logger.
+    """
+    # Import inside function to avoid circular dependency
+    from assisted_service_mcp.src.settings import settings
+
+    # Resolve logger name, falling back to a stable default
+    logger_name = settings.LOGGER_NAME or "assisted-service-mcp"
+    target_logger = logging.getLogger(logger_name)
+
+    # Configure third-party loggers
+    logging.getLogger("requests").setLevel(logging.ERROR)
+    urllib3_logger = logging.getLogger("urllib3")
+
+    # Reset handlers to prevent duplicates on reconfiguration
+    for handler in target_logger.handlers:
+        handler.close()
+    target_logger.handlers = []
+    for handler in urllib3_logger.handlers:
+        handler.close()
+    urllib3_logger.handlers = []
+
+    # Set levels
+    urllib3_logger.setLevel(logging.ERROR)
+    target_logger.setLevel(get_logging_level())
+
+    # Optional file logging
+    if settings.LOG_TO_FILE:
+        add_log_file_handler(target_logger, "assisted-service-mcp.log")
+        add_log_file_handler(urllib3_logger, "assisted-service-mcp.log")
+
+    # Always add stream handlers
+    add_stream_handler(target_logger)
+    add_stream_handler(urllib3_logger)
+
+    # Ensure modules using `from ...logger import log` get the configured logger
+    global log  # pylint: disable=global-statement
+    log = target_logger
+    return target_logger

assisted_service_mcp/src/main.py

@@ -0,0 +1,46 @@
+"""Main entry point for the Assisted Service MCP Server."""
+
+import uvicorn
+from assisted_service_mcp.src.api import app, server
+from assisted_service_mcp.src.settings import settings
+from assisted_service_mcp.src.metrics import metrics, initiate_metrics
+from assisted_service_mcp.src.logger import log
+
+
+def main() -> None:
+    """Start the MCP server.
+
+    Initializes the server, sets up metrics, and starts the uvicorn server.
+    """
+    try:
+        log.info("Starting Assisted Service MCP Server")
+        log.info(
+            "Configuration: TRANSPORT=%s, HOST=%s, PORT=%s",
+            settings.TRANSPORT,
+            settings.MCP_HOST,
+            settings.MCP_PORT,
+        )
+
+        # Initialize metrics with list of all tools
+        tool_names = server.list_tools_sync()
+        initiate_metrics(tool_names)
+        log.info("Initialized metrics for %s tools", len(tool_names))
+
+        # Add metrics endpoint
+        app.add_route("/metrics", metrics)
+        log.info("Metrics endpoint available at /metrics")
+
+        # Start the server using settings
+        uvicorn.run(app, host=settings.MCP_HOST, port=settings.MCP_PORT)
+
+    except KeyboardInterrupt:
+        log.info("Received keyboard interrupt, shutting down")
+    except Exception as e:
+        log.error("Server failed to start: %s", e, exc_info=True)
+        raise
+    finally:
+        log.info("Assisted Service MCP server shutting down")
+
+
+if __name__ == "__main__":
+    main()