list_streams
Retrieve available data streams in your OpenObserve organization to monitor logs and traces, with filtering and pagination options.
Instructions
List streams available in the current organization.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| stream_type | No | logs | |
| keyword | No | ||
| offset | No | ||
| limit | No | ||
| sort | No | name | |
| include_raw | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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, ) - 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, }, )