get_latest_traces
Retrieve recent trace data from OpenObserve streams using Unix timestamps to monitor application performance and debug issues.
Instructions
Get the latest trace data from a trace stream. Time values are Unix timestamps in microseconds.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| stream_name | Yes | ||
| start_time | Yes | ||
| end_time | Yes | ||
| size | No | ||
| offset | No | ||
| filter_query | No | ||
| timeout | No | ||
| include_raw | No |
Implementation Reference
- The actual implementation logic for the get_latest_traces tool in the client.
def get_latest_traces( self, *, stream_name: str, start_time: int, end_time: int, offset: int = 0, size: int = 20, filter_query: str | None = None, timeout: int | None = None, ) -> Any: query: dict[str, str | int | float | bool] = { "from": offset, "size": size, "start_time": start_time, "end_time": end_time, } if filter_query: query["filter"] = filter_query - openobserve_mcp/server.py:220-240 (registration)The tool registration and entry-point handler in the MCP server.
def get_latest_traces( stream_name: str, start_time: int, end_time: int, size: int = 20, offset: int = 0, filter_query: str | None = None, timeout: int | None = None, include_raw: bool = False, ) -> dict[str, Any]: """Get the latest trace data from a trace stream. Time values are Unix timestamps in microseconds.""" client = client_provider.get() raw = client.get_latest_traces( stream_name=stream_name, start_time=start_time, end_time=end_time, size=size, offset=offset, filter_query=filter_query, timeout=timeout, )