Fix search API bugs: timeZone defaulting, epoch time format, aggregate results, and circuit breaker semantics

sumologic_mcp/api_client.py

@@ -75,10 +75,17 @@ def __init__(self, config: SumoLogicConfig, auth: SumoLogicAuth):
             )
         )
 
+        # Only transient/infra-class failures should count toward opening the
+        # circuit. Bare `Exception` would mean user errors (e.g. invalid query
+        # syntax → 4xx, validation errors) also count, which can trip the
+        # breaker on benign batches of work and degrade tools unnecessarily.
         circuit_breaker_config = CircuitBreakerConfig(
             failure_threshold=max(5, self.config.max_retries * 2),
             recovery_timeout=60.0,
-            expected_exception=Exception,
+            expected_exception=(
+                APIError, RateLimitError, TimeoutError,
+                ConnectionError, OSError, httpx.RequestError,
+            ),
             success_threshold=3
         )
 
@@ -819,37 +826,56 @@ async def get_search_results(
                     search_state=job_state
                 )
 
-            # Get messages (log records)
+            # Try /records first (for aggregate queries), fall back to /messages.
+            # If both are empty, return empty results gracefully.
             params = {
                 "offset": offset,
                 "limit": limit
             }
-
-            response = await self._make_request(
-                method="GET",
-                endpoint=f"/api/v1/search/jobs/{job_id}/messages",
-                params=params,
-                operation_type="search_results"
-            )
-
-            results = await self._parse_json_response(response)
-
+
+            result_type = "records"
+            try:
+                response = await self._make_request(
+                    method="GET",
+                    endpoint=f"/api/v1/search/jobs/{job_id}/records",
+                    params=params,
+                    operation_type="search_results"
+                )
+                results = await self._parse_json_response(response)
+            except Exception:
+                results = {"records": [], "fields": []}
+
+            if not results.get("records"):
+                # No records — try messages (may also be empty for no-match queries)
+                result_type = "messages"
+                try:
+                    response = await self._make_request(
+                        method="GET",
+                        endpoint=f"/api/v1/search/jobs/{job_id}/messages",
+                        params=params,
+                        operation_type="search_results"
+                    )
+                    results = await self._parse_json_response(response)
+                except Exception:
+                    results = {"messages": [], "fields": []}
+
             logger.info(
                 f"Retrieved search results",
                 extra={
                     "job_id": job_id,
-                    "returned_count": len(results.get("messages", [])),
+                    "result_type": result_type,
+                    "returned_count": len(results.get(result_type, [])),
                     "offset": offset,
                     "limit": limit
                 }
             )
-            
+
             # Combine with job status for complete result
             return {
                 "job_id": job_id,
                 "status": status,
                 "results": results,
-                "messages": results.get("messages", []),
+                "messages": results.get(result_type, []),
                 "fields": results.get("fields", [])
             }
 
@@ -1542,14 +1568,14 @@ async def get_metric_metadata(
 
     async def list_collectors(self, limit: int = 100, offset: int = 0) -> Dict[str, Any]:
         """List collectors with pagination.
-        
+
         Args:
             limit: Maximum collectors to return (1-1000)
             offset: Result offset for pagination
-            
+
         Returns:
             Dictionary containing collectors list and metadata
-            
+
         Raises:
             ValidationError: If parameters are invalid
             APIError: If API request fails
@@ -1586,17 +1612,17 @@ async def list_collectors(self, limit: int = 100, offset: int = 0) -> Dict[str,
                 operation_type="collector"
             )
 
-            collectors = await self._parse_json_response(response)
-            
+            result = await self._parse_json_response(response)
+
             logger.info(
-                f"Retrieved {len(collectors.get('collectors', []))} collectors",
+                f"Retrieved {len(result.get('collectors', []))} collectors",
                 extra={
                     "limit": limit,
                     "offset": offset
                 }
             )
-            
-            return collectors
+
+            return result
 
         except APIError as e:
             raise APIError(

sumologic_mcp/api_validator.py

@@ -107,6 +107,7 @@ class SumoLogicAPIValidator:
         r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?([+-]\d{2}:\d{2}|Z)?$'
     )
     RELATIVE_TIME_PATTERN = re.compile(r'^-?\d+[smhdw]$|^now$')
+    EPOCH_PATTERN = re.compile(r'^\d{10,13}$')
 
     @classmethod
     def validate_search_params(cls, params: Dict[str, Any]) -> Dict[str, Any]:
@@ -135,8 +136,10 @@ def validate_search_params(cls, params: Dict[str, Any]) -> Dict[str, Any]:
                     api_endpoint="search API"
                 )
 
-        # Validate each parameter
+        # Validate each parameter (skip None so defaults get applied below)
         for param_name, param_value in params.items():
+            if param_value is None:
+                continue
             if param_name not in schema:
                 # Allow unknown parameters but log warning
                 validated_params[param_name] = param_value
@@ -334,8 +337,12 @@ def _validate_time_format(cls, param_name: str, time_value: str) -> None:
         # Check if it matches relative time format
         if cls.RELATIVE_TIME_PATTERN.match(time_value):
             return
-
-        # Neither format matched
+
+        # Check if it matches epoch time format (seconds or milliseconds)
+        if cls.EPOCH_PATTERN.match(time_value):
+            return
+
+        # No format matched
         raise TimeValidationError(
             f"Invalid time format for parameter '{param_name}'",
             time_value,

sumologic_mcp/resilience.py

@@ -9,7 +9,7 @@
 import time
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Callable, Dict, Optional, Type, Union, List
+from typing import Any, Callable, Dict, Optional, Tuple, Type, Union, List
 from datetime import datetime, timedelta
 
 from .exceptions import APIError, RateLimitError, TimeoutError, SumoLogicError
@@ -54,7 +54,10 @@ class CircuitBreakerConfig:
     """Configuration for circuit breaker."""
     failure_threshold: int = 5
     recovery_timeout: float = 60.0
-    expected_exception: Type[Exception] = Exception
+    # Pass a tuple of exception classes to count only transient/infra-class
+    # failures toward opening the circuit (the default Exception catches user
+    # errors too, which can trip the breaker on benign batches).
+    expected_exception: Union[Type[Exception], Tuple[Type[Exception], ...]] = Exception
     success_threshold: int = 3  # Successes needed in half-open to close
 
     def __post_init__(self):
@@ -221,7 +224,15 @@ async def _check_state_transition(self):
     async def _record_success(self):
         """Record successful operation."""
         self.stats.record_success()
-
+
+        # In CLOSED state, a success means the recent failure streak is broken.
+        # Without this reset, failure_count is monotonic and the breaker is
+        # guaranteed to trip eventually after enough cumulative failures over
+        # the lifetime of the process — "5 consecutive failures" is the
+        # intended semantics, not "5 failures ever".
+        if self.stats.state == CircuitState.CLOSED:
+            self.stats.failure_count = 0
+
         logger.debug(
             f"Circuit breaker '{self.name}' recorded success",
             extra={

sumologic_mcp/time_utils.py

@@ -201,30 +201,29 @@ def _parse_epoch_time(cls, time_str: str) -> datetime:
 
     @classmethod
     def to_sumo_api_format(cls, dt: datetime) -> str:
-        """Convert datetime to exact Sumo Logic API format.
-        
+        """Convert datetime to Sumo Logic search jobs API format.
+
         Args:
             dt: Datetime object to convert
-            
+
         Returns:
-            Time string in exact Sumo Logic API format (ISO 8601 with milliseconds and Z suffix)
+            Time string as epoch milliseconds (required by Sumo Logic search jobs API)
         """
-        # Sumo Logic API expects ISO 8601 format with milliseconds and Z suffix
-        # Format: YYYY-MM-DDTHH:MM:SS.sssZ
-        return dt.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
-    
+        import calendar
+        epoch_seconds = calendar.timegm(dt.timetuple())
+        return str(epoch_seconds * 1000 + dt.microsecond // 1000)
+
     @classmethod
     def to_sumo_time_format(cls, dt: datetime) -> str:
         """Convert datetime to Sumo Logic API time format.
-        
+
         Args:
             dt: Datetime object to convert
-            
+
         Returns:
-            Time string in Sumo Logic API format (ISO 8601 with milliseconds)
+            Time string as epoch milliseconds (required by Sumo Logic search jobs API)
         """
-        # Sumo Logic expects ISO format with milliseconds
-        return dt.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
+        return cls.to_sumo_api_format(dt)
 
     @classmethod
     def validate_time_range(cls, from_time: str, to_time: str) -> tuple[datetime, datetime]:

sumologic_mcp/tools/collector_tools.py

@@ -29,17 +29,15 @@ def __init__(self, api_client: SumoLogicAPIClient):
 
     async def list_collectors(
         self,
-        filter_type: Optional[str] = None,
         limit: int = 100,
         offset: int = 0
     ) -> Dict[str, Any]:
-        """List all collectors with optional filtering.
-        
+        """List all collectors with pagination.
+
         This tool retrieves a list of collectors from Sumo Logic with support
-        for type-based filtering and pagination.
-        
+        for pagination.
+
         Args:
-            filter_type: Optional collector type filter (Installable, Hosted, etc.)
             limit: Maximum number of collectors to return (1-1000)
             offset: Starting position for pagination (0-based)
 
@@ -66,15 +64,10 @@ async def list_collectors(
             if offset < 0:
                 raise ValidationError("Offset must be non-negative")
 
-            valid_types = ["Installable", "Hosted", "All"]
-            if filter_type and filter_type not in valid_types:
-                raise ValidationError(f"filter_type must be one of: {', '.join(valid_types)}")
-
-            logger.info(f"Listing collectors with type='{filter_type}', limit={limit}, offset={offset}")
-
+            logger.info(f"Listing collectors with limit={limit}, offset={offset}")
+
             # Get collectors from API
             collectors_response = await self.api_client.list_collectors(
-                filter_type=filter_type,
                 limit=limit,
                 offset=offset
             )
@@ -119,7 +112,7 @@ async def list_collectors(
                 "limit": limit,
                 "returned_count": len(formatted_collectors),
                 "has_more": has_more,
-                "filter_applied": filter_type is not None,
+                "filter_applied": False,
                 "collector_types": collector_types,
                 "summary": {
                     "online_collectors": sum(1 for c in formatted_collectors if c["status"]),
@@ -175,8 +168,13 @@ async def get_collector(self, collector_id: str) -> Dict[str, Any]:
             # Format collector data
             collector = collector_response.get("collector", collector_response)
 
-            # Get sources for this collector
-            sources = collector.get("sources", [])
+            # Get sources for this collector (requires separate API call)
+            try:
+                sources_response = await self.api_client.list_sources(collector_id)
+                sources = sources_response.get("sources", [])
+            except Exception as e:
+                logger.warning(f"Failed to fetch sources for collector {collector_id}: {e}")
+                sources = []
             formatted_sources = []
 
             for source in sources:
@@ -776,11 +774,6 @@ def get_tool_definitions(self) -> List[Dict[str, Any]]:
                 "inputSchema": {
                     "type": "object",
                     "properties": {
-                        "filter_type": {
-                            "type": "string",
-                            "description": "Optional collector type filter",
-                            "enum": ["Installable", "Hosted", "All"]
-                        },
                         "limit": {
                             "type": "integer",
                             "description": "Maximum number of collectors to return (1-1000)",