Performance testing

.env.example

@@ -510,13 +510,21 @@ RETRY_JITTER_MAX=0.5
 #####################################
 
 # Logging verbosity level
-# Options: DEBUG, INFO (default), WARNING, ERROR, CRITICAL
+# Options: DEBUG, INFO, WARNING, ERROR (default), CRITICAL
 # DEBUG: Detailed diagnostic info (verbose)
 # INFO: General operational messages
 # WARNING: Warning messages for potential issues
-# ERROR: Error messages for failures
+# ERROR: Error messages for failures (recommended for production)
 # CRITICAL: Only critical failures
-LOG_LEVEL=INFO
+# PRODUCTION: Use ERROR to minimize I/O overhead and improve performance
+LOG_LEVEL=ERROR
+
+# Disable access logging for performance
+# Options: true, false (default)
+# When true: Disables both gunicorn and uvicorn access logs
+# PRODUCTION: Set to true for high-performance deployments
+# Access logs create massive I/O overhead under high concurrency
+# DISABLE_ACCESS_LOG=true
 
 # Log output format
 # Options: json (default), text

docker-compose.yml

@@ -76,6 +76,13 @@ services:
       # Uncomment to enable catalog
       - MCPGATEWAY_CATALOG_ENABLED=true
       - MCPGATEWAY_CATALOG_FILE=/app/mcp-catalog.yml
+      # Authentication configuration
+      - AUTH_REQUIRED=true
+      - MCP_CLIENT_AUTH_ENABLED=true
+      - TRUST_PROXY_AUTH=false
+      # Logging configuration
+      - LOG_LEVEL=ERROR  # Default to ERROR for production performance
+      - DISABLE_ACCESS_LOG=true  # Disable uvicorn access logs for performance (massive I/O overhead)
 
       # Phoenix Observability Integration (uncomment when using Phoenix)
       # - PHOENIX_ENDPOINT=${PHOENIX_ENDPOINT:-http://phoenix:6006}

mcpgateway/admin.py

@@ -9910,7 +9910,7 @@ async def admin_generate_support_bundle(
         LOGGER.info(f"Support bundle generation requested by user: {user}")
 
         # First-Party
-        from mcpgateway.services.support_bundle_service import SupportBundleConfig, SupportBundleService
+        from mcpgateway.services.support_bundle_service import SupportBundleConfig, SupportBundleService  # pylint: disable=import-outside-toplevel
 
         # Create configuration
         config = SupportBundleConfig(

mcpgateway/config.py

@@ -626,7 +626,7 @@ def _parse_allowed_origins(cls, v):
         return set(v)
 
     # Logging
-    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(default="INFO", env="LOG_LEVEL")
+    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(default="ERROR", env="LOG_LEVEL")
     log_format: Literal["json", "text"] = "json"  # json or text
     log_to_file: bool = False  # Enable file logging (default: stdout/stderr only)
     log_filemode: str = "a+"  # append or overwrite

mcpgateway/main.py

@@ -3410,11 +3410,19 @@ async def handle_rpc(request: Request, db: Session = Depends(get_db), user=Depen
             request_id = params.get("requestId", None)
             if not uri:
                 raise JSONRPCError(-32602, "Missing resource URI in parameters", params)
-            result = await resource_service.read_resource(db, uri, request_id=request_id, user=get_user_email(user))
-            if hasattr(result, "model_dump"):
-                result = {"contents": [result.model_dump(by_alias=True, exclude_none=True)]}
-            else:
-                result = {"contents": [result]}
+            # Get user email for OAuth token selection
+            user_email = get_user_email(user)
+            try:
+                result = await resource_service.read_resource(db, uri, request_id=request_id, user=user_email)
+                if hasattr(result, "model_dump"):
+                    result = {"contents": [result.model_dump(by_alias=True, exclude_none=True)]}
+                else:
+                    result = {"contents": [result]}
+            except ValueError:
+                # Resource has no local content, forward to upstream MCP server
+                result = await gateway_service.forward_request(db, method, params, app_user_email=user_email)
+                if hasattr(result, "model_dump"):
+                    result = result.model_dump(by_alias=True, exclude_none=True)
         elif method == "prompts/list":
             if server_id:
                 prompts = await prompt_service.list_server_prompts(db, server_id, cursor=cursor)

mcpgateway/middleware/token_scoping.py

@@ -370,7 +370,10 @@ def _check_team_membership(self, payload: dict) -> bool:
 
         db = next(get_db())
         try:
-            for team_id in teams:
+            for team in teams:
+                # Extract team ID from dict or use string directly (backward compatibility)
+                team_id = team["id"] if isinstance(team, dict) else team
+
                 membership = db.execute(
                     select(EmailTeamMember).where(and_(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active))
                 ).scalar_one_or_none()
@@ -383,7 +386,7 @@ def _check_team_membership(self, payload: dict) -> bool:
         finally:
             db.close()
 
-    def _check_resource_team_ownership(self, request_path: str, token_teams: list) -> bool:
+    def _check_resource_team_ownership(self, request_path: str, token_teams: list) -> bool:  # pylint: disable=too-many-return-statements
         """
         Check if the requested resource is accessible by the token.
 
@@ -412,9 +415,16 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
         Returns:
             bool: True if resource access is allowed, False otherwise
         """
+        # Normalize token_teams: extract team IDs from dict objects (backward compatibility)
+        token_team_ids = []
+        for team in token_teams:
+            if isinstance(team, dict):
+                token_team_ids.append(team["id"])
+            else:
+                token_team_ids.append(team)
 
         # Determine token type
-        is_public_token = not token_teams or len(token_teams) == 0
+        is_public_token = not token_team_ids or len(token_team_ids) == 0
 
         if is_public_token:
             logger.debug("Processing request with PUBLIC-ONLY token")
@@ -442,7 +452,7 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
 
         # If no resource ID in path, allow (general endpoints like /health, /tokens, /metrics)
         if not resource_id or not resource_type:
-            logger.info(f"No resource ID found in path {request_path}, allowing access")
+            logger.debug(f"No resource ID found in path {request_path}, allowing access")
             return True
 
         # Import database models
@@ -477,16 +487,16 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
 
                 # TEAM-SCOPED SERVERS: Check if server belongs to token's teams
                 if server_visibility == "team":
-                    if server.team_id in token_teams:
+                    if server.team_id in token_team_ids:
                         logger.debug(f"Access granted: Team server {resource_id} belongs to token's team {server.team_id}")
                         return True
 
-                    logger.warning(f"Access denied: Server {resource_id} is team-scoped to '{server.team_id}', " f"token is scoped to teams {token_teams}")
+                    logger.warning(f"Access denied: Server {resource_id} is team-scoped to '{server.team_id}', " f"token is scoped to teams {token_team_ids}")
                     return False
 
                 # PRIVATE SERVERS: Check if server belongs to token's teams
                 if server_visibility == "private":
-                    if server.team_id in token_teams:
+                    if server.team_id in token_team_ids:
                         logger.debug(f"Access granted: Private server {resource_id} in token's team {server.team_id}")
                         return True
 
@@ -521,17 +531,17 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
                 # TEAM TOOLS: Check if tool's team matches token's teams
                 if tool_visibility == "team":
                     tool_team_id = getattr(tool, "team_id", None)
-                    if tool_team_id and tool_team_id in token_teams:
+                    if tool_team_id and tool_team_id in token_team_ids:
                         logger.debug(f"Access granted: Team tool {resource_id} belongs to token's team {tool_team_id}")
                         return True
 
-                    logger.warning(f"Access denied: Tool {resource_id} is team-scoped to '{tool_team_id}', " f"token is scoped to teams {token_teams}")
+                    logger.warning(f"Access denied: Tool {resource_id} is team-scoped to '{tool_team_id}', " f"token is scoped to teams {token_team_ids}")
                     return False
 
                 # PRIVATE TOOLS: Check if tool is in token's team context
                 if tool_visibility in ["private", "user"]:
                     tool_team_id = getattr(tool, "team_id", None)
-                    if tool_team_id and tool_team_id in token_teams:
+                    if tool_team_id and tool_team_id in token_team_ids:
                         logger.debug(f"Access granted: Private tool {resource_id} in token's team {tool_team_id}")
                         return True
 
@@ -566,17 +576,17 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
                 # TEAM RESOURCES: Check if resource's team matches token's teams
                 if resource_visibility == "team":
                     resource_team_id = getattr(resource, "team_id", None)
-                    if resource_team_id and resource_team_id in token_teams:
+                    if resource_team_id and resource_team_id in token_team_ids:
                         logger.debug(f"Access granted: Team resource {resource_id} belongs to token's team {resource_team_id}")
                         return True
 
-                    logger.warning(f"Access denied: Resource {resource_id} is team-scoped to '{resource_team_id}', " f"token is scoped to teams {token_teams}")
+                    logger.warning(f"Access denied: Resource {resource_id} is team-scoped to '{resource_team_id}', " f"token is scoped to teams {token_team_ids}")
                     return False
 
                 # PRIVATE RESOURCES: Check if resource is in token's team context
                 if resource_visibility in ["private", "user"]:
                     resource_team_id = getattr(resource, "team_id", None)
-                    if resource_team_id and resource_team_id in token_teams:
+                    if resource_team_id and resource_team_id in token_team_ids:
                         logger.debug(f"Access granted: Private resource {resource_id} in token's team {resource_team_id}")
                         return True
 
@@ -611,17 +621,17 @@ def _check_resource_team_ownership(self, request_path: str, token_teams: list) -
                 # TEAM PROMPTS: Check if prompt's team matches token's teams
                 if prompt_visibility == "team":
                     prompt_team_id = getattr(prompt, "team_id", None)
-                    if prompt_team_id and prompt_team_id in token_teams:
+                    if prompt_team_id and prompt_team_id in token_team_ids:
                         logger.debug(f"Access granted: Team prompt {resource_id} belongs to token's team {prompt_team_id}")
                         return True
 
-                    logger.warning(f"Access denied: Prompt {resource_id} is team-scoped to '{prompt_team_id}', " f"token is scoped to teams {token_teams}")
+                    logger.warning(f"Access denied: Prompt {resource_id} is team-scoped to '{prompt_team_id}', " f"token is scoped to teams {token_team_ids}")
                     return False
 
                 # PRIVATE PROMPTS: Check if prompt is in token's team context
                 if prompt_visibility in ["private", "user"]:
                     prompt_team_id = getattr(prompt, "team_id", None)
-                    if prompt_team_id and prompt_team_id in token_teams:
+                    if prompt_team_id and prompt_team_id in token_team_ids:
                         logger.debug(f"Access granted: Private prompt {resource_id} in token's team {prompt_team_id}")
                         return True
 

mcpgateway/services/logging_service.py

@@ -191,6 +191,9 @@ async def initialize(self) -> None:
             >>> service = LoggingService()
             >>> asyncio.run(service.initialize())
         """
+        # Update service log level from settings BEFORE configuring loggers
+        self._level = settings.log_level
+
         root_logger = logging.getLogger()
         self._loggers[""] = root_logger
 

mcpgateway/services/support_bundle_service.py

@@ -244,7 +244,7 @@ def _collect_system_info(self) -> Dict[str, Any]:
         # Try to collect psutil metrics if available
         try:
             # Third-Party
-            import psutil
+            import psutil  # pylint: disable=import-outside-toplevel
 
             info["system"] = {
                 "cpu_count": psutil.cpu_count(logical=True),
@@ -448,7 +448,7 @@ def generate_bundle(self, config: Optional[SupportBundleConfig] = None) -> Path:
                     zf.writestr(f"logs/{log_name}", log_content)
 
             # Add README
-            readme = """# MCP Gateway Support Bundle
+            readme = f"""# MCP Gateway Support Bundle
 
 This bundle contains diagnostic information for troubleshooting MCP Gateway issues.
 
@@ -478,12 +478,10 @@ def generate_bundle(self, config: Optional[SupportBundleConfig] = None) -> Path:
 Pay special attention to logs/ for error messages and stack traces.
 
 ---
-Generated: {timestamp}
-Hostname: {hostname}
-Version: {version}
-""".format(
-                timestamp=self.timestamp.isoformat(), hostname=self.hostname, version=__version__
-            )
+Generated: {self.timestamp.isoformat()}
+Hostname: {self.hostname}
+Version: {__version__}
+"""
 
             zf.writestr("README.md", readme)