Bug fixes

Signed-off-by: Madhav Kandukuri <[email protected]>

Author: madhav165

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

@@ -113,6 +113,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/main.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()
@@ -2019,7 +2028,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/services/gateway_service.py

@@ -23,7 +23,6 @@
 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,13 +136,17 @@ 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=tool.url,
+                    url=str(gateway.url),
                     description=tool.description,
                     integration_type=tool.integration_type,
                     request_type=tool.request_type,
@@ -157,6 +159,9 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
                 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,
@@ -166,7 +171,8 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
                 last_seen=datetime.now(timezone.utc),
                 auth_type=auth_type,
                 auth_value=auth_value,
-                tools=tools,
+                tools=new_tools,
+                # federated_tools=existing_tools + new_tools
             )
 
             # Add to DB
@@ -181,12 +187,19 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             await self._notify_gateway_added(db_gateway)
 
             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.
@@ -462,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()
+
+                    # Mark successful check
+                    gateway.last_seen = datetime.utcnow()
 
-                # Update last seen
-                gateway.last_seen = datetime.utcnow()
-                return True
+                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.
@@ -584,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()
 
@@ -598,7 +629,6 @@ async def _run_health_checks(self) -> None:
                 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/tool_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:

tests/hey/payload2.json

@@ -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"
+    }
+  }
+}