get_error_logs

Implementation Reference

• mcp_graylog/server.py:83-98 (handler)

  The main handler implementation of get_error_logs. It builds a MessageSearchInput with a query for ERROR/CRITICAL/FATAL levels, using a relative time range, and delegates to graylog.search_messages.

  def get_error_logs(
      self,
      hours: Annotated[int, Field(ge=1)] = 1,
      limit: Annotated[int, Field(ge=1, le=1000)] = 100,
  ) -> dict[str, Any]:
      if hours < 1:
          raise ValueError("hours must be at least 1")
  
      search = MessageSearchInput(
          query="level:ERROR OR level:CRITICAL OR level:FATAL",
          timerange=RelativeTimeRange(value=hours, unit="h"),
          fields=["timestamp", "source", "level", "message"],
          limit=limit,
          offset=0,
      )
      return self.graylog.search_messages(search)

• mcp_graylog/models.py:44-54 (schema)

  MessageSearchInput Pydantic model used by get_error_logs for input validation. Fields include query, timerange, fields, limit, offset.

  class MessageSearchInput(BaseModel):
      query: str = Field("*", min_length=1)
      timerange: TimeRange = Field(
          default_factory=lambda: RelativeTimeRange.model_validate({})
      )
      streams: list[str] = Field(default_factory=list)
      fields: list[str] = Field(
          default_factory=lambda: ["timestamp", "source", "level", "message"]
      )
      limit: int = Field(50, ge=1, le=1000)
      offset: int = Field(0, ge=0)

• mcp_graylog/models.py:8-19 (schema)

  RelativeTimeRange Pydantic model used by get_error_logs to specify the time range (hours back).

  class RelativeTimeRange(BaseModel):
      model_config = ConfigDict(extra="forbid")
  
      value: int = Field(1, ge=1)
      unit: Literal["s", "m", "h", "d", "w"] = Field("h")
  
      def to_seconds(self) -> int:
          multipliers = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}
          return self.value * multipliers[self.unit]
  
      def to_graylog(self) -> dict[str, int | str]:
          return {"type": "relative", "range": self.to_seconds()}

• mcp_graylog/server.py:135-135 (registration)

  Registers get_error_logs as an MCP tool via mcp.tool()(handlers.get_error_logs).

  mcp.tool()(handlers.get_error_logs)

• mcp_graylog/server.py:83-98 (helper)

  ToolHandlers dataclass that holds a reference to the GraylogToolClient protocol; get_error_logs is a method on it.

  def get_error_logs(
      self,
      hours: Annotated[int, Field(ge=1)] = 1,
      limit: Annotated[int, Field(ge=1, le=1000)] = 100,
  ) -> dict[str, Any]:
      if hours < 1:
          raise ValueError("hours must be at least 1")
  
      search = MessageSearchInput(
          query="level:ERROR OR level:CRITICAL OR level:FATAL",
          timerange=RelativeTimeRange(value=hours, unit="h"),
          fields=["timestamp", "source", "level", "message"],
          limit=limit,
          offset=0,
      )
      return self.graylog.search_messages(search)