search_logs

Implementation Reference

• mcp_graylog/server.py:30-31 (handler)

  The search_logs handler function in the ToolHandlers dataclass. It receives a MessageSearchInput and delegates to graylog.search_messages().

  def search_logs(self, search: MessageSearchInput) -> dict[str, Any]:
      return self.graylog.search_messages(search)

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

  The MessageSearchInput Pydantic model used as the input schema for search_logs, defining fields like query, timerange, streams, 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/server.py:128-128 (registration)

  Registration of search_logs as an MCP tool via FastMCP.tool() decorator pattern.

  mcp.tool()(handlers.search_logs)

• mcp_graylog/models.py:64-65 (helper)

  The to_graylog_payload() method on MessageSearchInput converts the typed input to the Graylog API JSON payload format.

  def to_graylog_payload(self) -> dict[str, object]:
      payload: dict[str, object] = {

• mcp_graylog/graylog_client.py:60-65 (helper)

  The actual HTTP call that search_logs delegates to: GraylogClient.search_messages() posts to /api/search/messages.

  def search_messages(self, search: MessageSearchInput) -> dict[str, Any]:
      return self._request(
          "POST",
          "/api/search/messages",
          json=search.to_graylog_payload(),
      )