Merge pull request #72 from IBM/delete-gateway-fix

Fix gateway deletion and addition

Author: crivetimihai

mcpgateway/cli.py

@@ -60,13 +60,26 @@ def _needs_app(arg_list: List[str]) -> bool:
     is taken as the application path. We therefore look at the first
     element of *arg_list* (if any) – if it *starts* with a dash it must be
     an option, hence the app path is missing and we should inject ours.
+
+    Args:
+        arg_list (List[str]): List of arguments
+
+    Returns:
+        bool: Returns *True* when the CLI invocation has *no* positional APP path
     """
 
     return len(arg_list) == 0 or arg_list[0].startswith("-")
 
 
 def _insert_defaults(raw_args: List[str]) -> List[str]:
-    """Return a *new* argv with defaults sprinkled in where needed."""
+    """Return a *new* argv with defaults sprinkled in where needed.
+
+    Args:
+        raw_args (List[str]): List of input arguments to cli
+
+    Returns:
+        List[str]: List of arguments
+    """
 
     args = list(raw_args)  # shallow copy – we'll mutate this
 

mcpgateway/config.py

@@ -122,6 +122,9 @@ def _parse_allowed_origins(cls, v):
     # For federation_peers strip out quotes to ensure we're passing valid JSON via env
     federation_peers: Annotated[List[str], NoDecode] = []
 
+    # Lock file path for initializing gateway service initialize
+    lock_file_path: str = "/tmp/gateway_init.done"
+
     @field_validator("federation_peers", mode="before")
     @classmethod
     def _parse_federation_peers(cls, v):

mcpgateway/db.py

@@ -28,6 +28,7 @@
 import asyncio
 import json
 import logging
+import os
 from contextlib import asynccontextmanager
 from typing import Any, AsyncIterator, Dict, List, Optional, Union
 
@@ -170,7 +171,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
     shuts them down in reverse order on exit.
 
     Args:
-        app (FastAPI): FastAPI app
+        _app (FastAPI): FastAPI app
 
     Yields:
         None
@@ -184,7 +185,15 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         await tool_service.initialize()
         await resource_service.initialize()
         await prompt_service.initialize()
-        await gateway_service.initialize()
+        try:
+            # Try to create the file exclusively
+            fd = os.open(settings.lock_file_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        except FileExistsError:
+            logger.info("Gateway already initialized by another worker")
+        else:
+            with os.fdopen(fd, "w") as lock_file:
+                lock_file.write("initialized")
+            await gateway_service.initialize()
         await root_service.initialize()
         await completion_service.initialize()
         await logging_service.initialize()
@@ -2052,7 +2061,9 @@ async def root_redirect(request: Request):
             RedirectResponse: Redirects to /admin.
         """
         logger.debug("Redirecting root path to /admin")
-        return RedirectResponse(request.url_for("admin_home"))
+        root_path = request.scope.get("root_path", "")
+        return RedirectResponse(f"{root_path}/admin", status_code=303)
+        # return RedirectResponse(request.url_for("admin_home"))
 
 else:
     # If UI is disabled, provide API info at root

mcpgateway/main.py

@@ -402,7 +402,7 @@ class ToolRead(BaseModelWithConfig):
 
     id: int
     name: str
-    url: str
+    url: Optional[str]
     description: Optional[str]
     request_type: str
     integration_type: str

mcpgateway/schemas.py

@@ -16,14 +16,13 @@
 
 import asyncio
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, AsyncGenerator, Dict, List, Optional, Set
 
 import httpx
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from mcpgateway.config import settings
@@ -122,7 +121,6 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
 
         Raises:
             GatewayNameConflictError: If gateway name already exists
-            GatewayError: If registration fails
         """
         try:
             # Check for name conflicts (both active and inactive)
@@ -138,12 +136,43 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             auth_type = getattr(gateway, "auth_type", None)
             auth_value = getattr(gateway, "auth_value", {})
 
-            # Initialize connection and get capabilities
             capabilities, tools = await self._initialize_gateway(str(gateway.url), auth_value)
 
+            all_names = [td.name for td in tools]
+
+            existing_tools = db.execute(select(DbTool).where(DbTool.name.in_(all_names))).scalars().all()
+            existing_tool_names = [tool.name for tool in existing_tools]
+
+            tools = [
+                DbTool(
+                    name=tool.name,
+                    url=str(gateway.url),
+                    description=tool.description,
+                    integration_type=tool.integration_type,
+                    request_type=tool.request_type,
+                    headers=tool.headers,
+                    input_schema=tool.input_schema,
+                    jsonpath_filter=tool.jsonpath_filter,
+                    auth_type=auth_type,
+                    auth_value=auth_value,
+                )
+                for tool in tools
+            ]
+
+            existing_tools = [tool for tool in tools if tool.name in existing_tool_names]
+            new_tools = [tool for tool in tools if tool.name not in existing_tool_names]
+
             # Create DB model
             db_gateway = DbGateway(
-                name=gateway.name, url=str(gateway.url), description=gateway.description, capabilities=capabilities, last_seen=datetime.utcnow(), auth_type=auth_type, auth_value=auth_value
+                name=gateway.name,
+                url=str(gateway.url),
+                description=gateway.description,
+                capabilities=capabilities,
+                last_seen=datetime.now(timezone.utc),
+                auth_type=auth_type,
+                auth_value=auth_value,
+                tools=new_tools,
+                # federated_tools=existing_tools + new_tools
             )
 
             # Add to DB
@@ -157,23 +186,20 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             # Notify subscribers
             await self._notify_gateway_added(db_gateway)
 
-            inserted_gateway = db.execute(select(DbGateway).where(DbGateway.name == gateway.name)).scalar_one_or_none()
-            inserted_gateway_id = inserted_gateway.id
-
-            logger.info(f"Registered gateway: {gateway.name}")
-
-            for tool in tools:
-                tool.gateway_id = inserted_gateway_id
-                await self.tool_service.register_tool(db=db, tool=tool)
-
             return GatewayRead.model_validate(gateway)
-
-        except IntegrityError:
-            db.rollback()
-            raise GatewayError(f"Gateway already exists: {gateway.name}")
-        except Exception as e:
-            db.rollback()
-            raise GatewayError(f"Failed to register gateway: {str(e)}")
+        except* ValueError as ve:
+            logger.error("ValueErrors in group: %s", ve.exceptions)
+        except* RuntimeError as re:
+            logger.error("RuntimeErrors in group: %s", re.exceptions)
+        except* BaseException as other:  # catches every other sub-exception
+            logger.error("Other grouped errors: %s", other.exceptions)
+        # except IntegrityError as ex:
+        #     logger.error(f"Error adding gateway: {ex}")
+        #     db.rollback()
+        #     raise GatewayError(f"Gateway already exists: {gateway.name}")
+        # except Exception as e:
+        #     db.rollback()
+        #     raise GatewayError(f"Failed to register gateway: {str(e)}")
 
     async def list_gateways(self, db: Session, include_inactive: bool = False) -> List[GatewayRead]:
         """List all registered gateways.
@@ -393,14 +419,6 @@ async def delete_gateway(self, db: Session, gateway_id: int) -> None:
             # Store gateway info for notification before deletion
             gateway_info = {"id": gateway.id, "name": gateway.name, "url": gateway.url}
 
-            # Remove associated tools
-            try:
-                db.query(DbTool).filter(DbTool.gateway_id == gateway_id).delete()
-                db.commit()
-                logger.info(f"Deleted tools associated with gateway: {gateway.name}")
-            except Exception as ex:
-                logger.warning(f"No tools found: {ex}")
-
             # Hard delete gateway
             db.delete(gateway)
             db.commit()
@@ -457,28 +475,42 @@ async def forward_request(self, gateway: DbGateway, method: str, params: Optiona
             raise GatewayConnectionError(f"Failed to forward request to {gateway.name}: {str(e)}")
 
     async def check_health_of_gateways(self, gateways: List[DbGateway]) -> bool:
-        """Health check for gateways
+        """Health check for a list of gateways.
+
+        Deactivates gateway if gateway is not healthy.
 
         Args:
-            gateways: Gateways to check
+            gateways (List[DbGateway]): List of gateways to check if healthy
 
         Returns:
-            True if gateway is healthy
+            bool: True if all  active gateways are healthy
         """
-        for gateway in gateways:
-            if not gateway.is_active:
-                return False
+        # Reuse a single HTTP client for all requests
+        async with httpx.AsyncClient() as client:
+            for gateway in gateways:
+                # Inactive gateways are unhealthy
+                if not gateway.is_active:
+                    continue
 
-            try:
-                # Try to initialize connection
-                await self._initialize_gateway(gateway.url, gateway.auth_value)
+                try:
+                    # Ensure auth_value is a dict
+                    auth_data = gateway.auth_value or {}
+                    headers = decode_auth(auth_data)
+
+                    # Perform the GET and raise on 4xx/5xx
+                    async with client.stream("GET", gateway.url, headers=headers) as response:
+                        # This will raise immediately if status is 4xx/5xx
+                        response.raise_for_status()
 
-                # Update last seen
-                gateway.last_seen = datetime.utcnow()
-                return True
+                    # Mark successful check
+                    gateway.last_seen = datetime.utcnow()
+
+                except Exception:
+                    with SessionLocal() as db:
+                        await self.toggle_gateway_status(db=db, gateway_id=gateway.id, activate=False)
 
-            except Exception:
-                return False
+        # All gateways passed
+        return True
 
     async def aggregate_capabilities(self, db: Session) -> Dict[str, Any]:
         """Aggregate capabilities from all gateways.
@@ -579,7 +611,11 @@ async def connect_to_sse_server(server_url: str, authentication: Optional[Dict[s
             raise GatewayConnectionError(f"Failed to initialize gateway at {url}: {str(e)}")
 
     def _get_active_gateways(self) -> list[DbGateway]:
-        """Sync function for database operations (runs in thread)."""
+        """Sync function for database operations (runs in thread).
+
+        Returns:
+            List[DbGateway]: List of active gateways
+        """
         with SessionLocal() as db:
             return db.execute(select(DbGateway).where(DbGateway.is_active)).scalars().all()
 
@@ -590,9 +626,9 @@ async def _run_health_checks(self) -> None:
                 # Run sync database code in a thread
                 gateways = await asyncio.to_thread(self._get_active_gateways)
 
-                # Async health checks (non-blocking)
-                await self.check_health_of_gateways(gateways)
-
+                if len(gateways) > 0:
+                    # Async health checks (non-blocking)
+                    await self.check_health_of_gateways(gateways)
             except Exception as e:
                 logger.error(f"Health check run failed: {str(e)}")
 

mcpgateway/services/gateway_service.py

@@ -518,12 +518,15 @@ async def invoke_tool(self, db: Session, name: str, arguments: Dict[str, Any]) -
                 else:
                     headers = {}
 
-                async def connect_to_sse_server(server_url: str):
+                async def connect_to_sse_server(server_url: str) -> str:
                     """
                     Connect to an MCP server running with SSE transport
 
                     Args:
                         server_url (str): MCP Server SSE URL
+
+                    Returns:
+                        str: Result of tool call
                     """
                     # Use async with directly to manage the context
                     async with sse_client(url=server_url, headers=headers) as streams:

mcpgateway/services/tool_service.py

@@ -0,0 +1,13 @@
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "method": "tools/call",
+  "params": {
+    "name": "convert_time",
+    "arguments": {
+      "source_timezone": "Europe/Berlin",
+      "target_timezone": "Europe/Dublin",
+      "time": "09:00"
+    }
+  }
+}