feat: gate HTTP streaming mode using IDCS

src/oci-api-mcp-server/oracle/oci_api_mcp_server/__init__.py

@@ -5,4 +5,4 @@
 """
 
 __project__ = "oracle.oci-api-mcp-server"
-__version__ = "1.1.4"
+__version__ = "2.0.0"

src/oci-api-mcp-server/oracle/oci_api_mcp_server/server.py

@@ -18,8 +18,6 @@
 
 logger = Logger(__project__, level="INFO")
 
-mcp = FastMCP(name=__project__)
-
 # setup user agent
 user_agent_name = __project__.split("oracle.", 1)[1].split("-server", 1)[0]
 USER_AGENT = f"{user_agent_name}/{__version__}"
@@ -52,7 +50,6 @@ def get_oci_commands() -> str:
     env_copy["OCI_SDK_APPEND_USER_AGENT"] = USER_AGENT
 
     try:
-        # Run OCI CLI command using subprocess
         result = subprocess.run(
             ["oci", "--help"],
             env=env_copy,
@@ -112,7 +109,6 @@ def get_oci_command_help(command: str) -> str:
     env_copy["OCI_SDK_APPEND_USER_AGENT"] = USER_AGENT
 
     try:
-        # Run OCI CLI command using subprocess
         result = subprocess.run(
             ["oci"] + command.split() + ["--help"],
             env=env_copy,
@@ -154,7 +150,6 @@ def run_oci_command(
     profile = os.getenv("OCI_CONFIG_PROFILE", oci.config.DEFAULT_PROFILE)
     logger.info(f"run_oci_command called with command: {command} --profile {profile}")
 
-    # Check if command is in denylist
     if denylist_manager.isCommandInDenyList(command):
         error_message = (
             f"Command '{command}' is denied by denylist. This command is found in the "
@@ -166,10 +161,9 @@ def run_oci_command(
         logger.error(error_message)
         return {"error": error_message}
 
-    # Run OCI CLI command using subprocess
     try:
         result = subprocess.run(
-            ["oci", "--profile"] + [profile] + ["--auth", "security_token"] + command.split(),
+            ["oci", "--profile", profile, "--auth", "security_token"] + command.split(),
             env=env_copy,
             capture_output=True,
             text=True,
@@ -188,9 +182,7 @@ def run_oci_command(
 
         try:
             response["output"] = json.loads(result.stdout)
-        except TypeError:
-            pass
-        except json.JSONDecodeError:
+        except (TypeError, json.JSONDecodeError):
             pass
 
         return response
@@ -204,14 +196,9 @@ def run_oci_command(
 
 
 def main():
-
-    host = os.getenv("ORACLE_MCP_HOST")
-    port = os.getenv("ORACLE_MCP_PORT")
-
-    if host and port:
-        mcp.run(transport="http", host=host, port=int(port))
-    else:
-        mcp.run()
+    if os.getenv("ORACLE_MCP_HOST") or os.getenv("ORACLE_MCP_PORT"):
+        raise RuntimeError("oracle.oci-api-mcp-server supports stdio transport only.")
+    mcp.run()
 
 
 if __name__ == "__main__":

src/oci-api-mcp-server/oracle/oci_api_mcp_server/tests/test_oci_api_tools.py

@@ -11,6 +11,7 @@
 
 import pytest
 from fastmcp import Client
+import oracle.oci_api_mcp_server.server as server
 from oracle.oci_api_mcp_server import __project__
 from oracle.oci_api_mcp_server.server import mcp
 
@@ -192,53 +193,31 @@ async def test_run_oci_command_denied(self, mock_json_loads, mock_run):
 
 
 class TestServer:
-    @patch("oracle.oci_api_mcp_server.server.mcp.run")
-    @patch("os.getenv")
-    def test_main_with_host_and_port(self, mock_getenv, mock_mcp_run):
-        mock_env = {
-            "ORACLE_MCP_HOST": "1.2.3.4",
-            "ORACLE_MCP_PORT": 8888,
-        }
-
-        mock_getenv.side_effect = lambda x: mock_env.get(x)
-        import oracle.oci_api_mcp_server.server as server
-
-        server.main()
-        mock_mcp_run.assert_called_once_with(
-            transport="http",
-            host=mock_env["ORACLE_MCP_HOST"],
-            port=mock_env["ORACLE_MCP_PORT"],
-        )
-
     @patch("oracle.oci_api_mcp_server.server.mcp.run")
     @patch("os.getenv")
     def test_main_without_host_and_port(self, mock_getenv, mock_mcp_run):
         mock_getenv.return_value = None
-        import oracle.oci_api_mcp_server.server as server
 
         server.main()
         mock_mcp_run.assert_called_once_with()
 
-    @patch("oracle.oci_api_mcp_server.server.mcp.run")
     @patch("os.getenv")
-    def test_main_with_only_host(self, mock_getenv, mock_mcp_run):
-        mock_env = {
-            "ORACLE_MCP_HOST": "1.2.3.4",
-        }
-        mock_getenv.side_effect = lambda x: mock_env.get(x)
-        import oracle.oci_api_mcp_server.server as server
+    def test_main_with_host_and_port(self, mock_getenv):
+        mock_getenv.side_effect = lambda x: {"ORACLE_MCP_HOST": "1.2.3.4", "ORACLE_MCP_PORT": "8888"}.get(x)
 
-        server.main()
-        mock_mcp_run.assert_called_once_with()
+        with pytest.raises(RuntimeError, match="stdio transport only"):
+            server.main()
 
-    @patch("oracle.oci_api_mcp_server.server.mcp.run")
     @patch("os.getenv")
-    def test_main_with_only_port(self, mock_getenv, mock_mcp_run):
-        mock_env = {
-            "ORACLE_MCP_PORT": "8888",
-        }
-        mock_getenv.side_effect = lambda x: mock_env.get(x)
-        import oracle.oci_api_mcp_server.server as server
+    def test_main_with_only_host(self, mock_getenv):
+        mock_getenv.side_effect = lambda x: {"ORACLE_MCP_HOST": "1.2.3.4"}.get(x)
 
-        server.main()
-        mock_mcp_run.assert_called_once_with()
+        with pytest.raises(RuntimeError, match="stdio transport only"):
+            server.main()
+
+    @patch("os.getenv")
+    def test_main_with_only_port(self, mock_getenv):
+        mock_getenv.side_effect = lambda x: {"ORACLE_MCP_PORT": "8888"}.get(x)
+
+        with pytest.raises(RuntimeError, match="stdio transport only"):
+            server.main()

src/oci-api-mcp-server/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oracle.oci-api-mcp-server"
-version = "1.1.4"
+version = "2.0.0"
 description = "OCI CLI MCP server"
 readme = "README.md"
 requires-python = ">=3.13"

src/oci-cloud-guard-mcp-server/oracle/oci_cloud_guard_mcp_server/__init__.py

@@ -5,4 +5,4 @@
 """
 
 __project__ = "oracle.oci-cloud-guard-mcp-server"
-__version__ = "1.1.5"
+__version__ = "2.0.0"

src/oci-cloud-guard-mcp-server/oracle/oci_cloud_guard_mcp_server/server.py

@@ -4,13 +4,16 @@
 https://oss.oracle.com/licenses/upl.
 """
 
+import configparser
 import os
 from datetime import datetime, timedelta, timezone
 from logging import Logger
 from typing import Literal, Optional
 
 import oci
 from fastmcp import FastMCP
+from fastmcp.server.auth.providers.oci import OCIProvider
+from fastmcp.server.dependencies import get_access_token
 from oci.cloud_guard import CloudGuardClient
 from oracle.oci_cloud_guard_mcp_server.models import (
     Problem,
@@ -25,6 +28,40 @@
 mcp = FastMCP(name=__project__)
 
 
+def _get_profile_value(key: str):
+    parser = configparser.ConfigParser()
+    parser.read(os.path.expanduser(os.getenv("OCI_CONFIG_FILE", oci.config.DEFAULT_LOCATION)))
+    profile = os.getenv("OCI_CONFIG_PROFILE", oci.config.DEFAULT_PROFILE)
+    return (parser[profile].get(key) if profile in parser else None) or parser.defaults().get(key)
+
+
+def _get_http_config_and_signer():
+    if not (os.getenv("ORACLE_MCP_HOST") and os.getenv("ORACLE_MCP_PORT")):
+        return None, None
+    token = get_access_token()
+    if token is None:
+        raise RuntimeError("HTTP requests require an authenticated IDCS access token.")
+    domain = os.getenv("IDCS_DOMAIN")
+    client_id = os.getenv("IDCS_CLIENT_ID")
+    client_secret = os.getenv("IDCS_CLIENT_SECRET")
+    if not all((domain, client_id, client_secret)):
+        raise RuntimeError(
+            "HTTP requests require IDCS authentication. Set IDCS_DOMAIN, IDCS_CLIENT_ID, and IDCS_CLIENT_SECRET."
+        )
+    region = os.getenv("OCI_REGION") or _get_profile_value("region")
+    if not region:
+        raise RuntimeError("HTTP requests require OCI_REGION or an OCI config file region.")
+    config = {"region": region}
+    user_agent_name = __project__.split("oracle.", 1)[1].split("-server", 1)[0]
+    config["additional_user_agent"] = f"{user_agent_name}/{__version__}"
+    return config, oci.auth.signers.TokenExchangeSigner(
+        token.token,
+        f"https://{domain}",
+        client_id,
+        client_secret,
+        region=config.get("region"),
+    )
+
 def _get_oci_client_kwargs(signer=None):
     kwargs = {
         "circuit_breaker_strategy": oci.circuit_breaker.CircuitBreakerStrategy(
@@ -41,6 +78,9 @@ def _get_oci_client_kwargs(signer=None):
 
 
 def get_cloud_guard_client():
+    config, signer = _get_http_config_and_signer()
+    if signer is not None:
+        return CloudGuardClient(config, **_get_oci_client_kwargs(signer))
     config = oci.config.from_file(
         file_location=os.getenv("OCI_CONFIG_FILE", oci.config.DEFAULT_LOCATION),
         profile_name=os.getenv("OCI_CONFIG_PROFILE", oci.config.DEFAULT_PROFILE),
@@ -139,10 +179,24 @@ def main():
     host = os.getenv("ORACLE_MCP_HOST")
     port = os.getenv("ORACLE_MCP_PORT")
 
-    if host and port:
-        mcp.run(transport="http", host=host, port=int(port))
-    else:
+    if not (host and port):
         mcp.run()
+        return
+    domain = os.getenv("IDCS_DOMAIN")
+    client_id = os.getenv("IDCS_CLIENT_ID")
+    client_secret = os.getenv("IDCS_CLIENT_SECRET")
+    if not all((domain, client_id, client_secret)):
+        raise RuntimeError(
+            "HTTP transport requires IDCS authentication. "
+            "Set IDCS_DOMAIN, IDCS_CLIENT_ID, IDCS_CLIENT_SECRET, ORACLE_MCP_HOST, and ORACLE_MCP_PORT."
+        )
+    mcp.auth = OCIProvider(
+        config_url=f"https://{domain}/.well-known/openid-configuration",
+        client_id=client_id,
+        client_secret=client_secret,
+        base_url=f"http://{host}:{port}",
+    )
+    mcp.run(transport="http", host=host, port=int(port))
 
 
 if __name__ == "__main__":