openobserve-community-mcp

Instructions

Implementation Reference

• openobserve_mcp/server.py:49-72 (handler)

  The MCP tool definition for `list_streams` in `openobserve_mcp/server.py`.

  @server.tool()
  def list_streams(
      stream_type: str = "logs",
      keyword: str = "",
      offset: int = 0,
      limit: int = 50,
      sort: str = "name",
      include_raw: bool = False,
  ) -> dict[str, Any]:
      """List streams available in the current organization."""
      client = client_provider.get()
      raw = client.list_streams(
          stream_type=stream_type,
          keyword=keyword,
          offset=offset,
          limit=limit,
          sort=sort,
      )
      return build_list_streams_result(
          org_id=client.resolve_org_id(),
          stream_type=stream_type,
          raw=raw,
          include_raw=include_raw,
      )

• openobserve_mcp/openobserve_client.py:116-135 (handler)

  The underlying OpenObserve API client method that performs the network request for `list_streams`.

  def list_streams(
      self,
      *,
      stream_type: str,
      keyword: str = "",
      offset: int = 0,
      limit: int = 50,
      sort: str = "name",
  ) -> Any:
      return self.request_json(
          "GET",
          self._org_path("/api/{org_id}/streams"),
          query={
              "type": stream_type,
              "keyword": keyword,
              "offset": offset,
              "limit": limit,
              "sort": sort,
          },
      )