feat: [ECO-515] - Add traces to MCP server (#215)

* feat: [ECO-515] - Add traces to MCP server

* fix: add tests & improve search history

* fix: tests

* fix: format

Author: YassineGabsi

src/deepset_mcp/api/search_history/__init__.py

@@ -4,7 +4,7 @@
 
 """Search history API module."""
 
-from .models import SearchHistoryEntry
+from .models import HaystackTraceV1, PipelineTraceEntry, SearchHistoryEntry
 from .resource import SearchHistoryResource
 
-__all__ = ["SearchHistoryEntry", "SearchHistoryResource"]
+__all__ = ["HaystackTraceV1", "PipelineTraceEntry", "SearchHistoryEntry", "SearchHistoryResource"]

src/deepset_mcp/api/search_history/models.py

@@ -6,22 +6,77 @@
 
 from typing import Any
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class SearchHistoryEntry(BaseModel):
-    """A single search history entry from the deepset platform.
+    """A single search history entry from the deepset platform (v1 API format).
 
-    Contains query, answers, prompts, feedback, and other metadata.
+    The v1 search history API returns entries with the following top-level fields.
+    Key nested data:
+    - The search query text is at ``request.query``.
+    - The timestamp is at ``time`` (also aliased to ``created_at`` for convenience).
+    - The pipeline name is at ``pipeline.name``.
+    - Search results are in ``response`` (list of result entries).
     """
 
     model_config = {"extra": "allow"}
 
-    query: str | None = Field(default=None, description="The search query that was executed")
-    answer: str | None = Field(default=None, description="The answer returned by the pipeline")
-    created_at: str | None = Field(default=None, description="When the search was performed")
-    pipeline_name: str | None = Field(default=None, description="Name of the pipeline used")
-    feedback: list[dict[str, Any]] | None = Field(default=None, description="User feedback on the search")
+    # -- Record identifiers --
+    search_history_id: str | None = Field(default=None, description="Unique identifier for this search history record")
+    session_id: str | None = Field(default=None, description="Session identifier grouping related searches")
+
+    # -- The search request --
+    request: dict[str, Any] | None = Field(
+        default=None,
+        description="The original search request (contains 'query', 'filters', 'params', etc.)",
+    )
+    # Convenience field — extracted from request.query via model_validator
+    query: str | None = Field(default=None, description="The search query text (extracted from request.query)")
+
+    # -- The search response --
+    response: list[dict[str, Any]] | None = Field(
+        default=None, description="List of search result entries returned by the pipeline"
+    )
+
+    # -- Timing & status --
+    time: str | None = Field(default=None, description="ISO-8601 timestamp when the search was performed")
+    # Convenience alias — populated from `time` via model_validator
+    created_at: str | None = Field(default=None, description="Alias for `time` — when the search was performed")
+    duration: float | None = Field(default=None, description="End-to-end query duration in seconds")
+    status: str | None = Field(default=None, description="Run status: 'success' or 'failed'")
+
+    # -- Pipeline & version --
+    pipeline: dict[str, Any] | None = Field(
+        default=None, description="Pipeline metadata (contains 'name' and other pipeline info)"
+    )
+    pipeline_version_id: str | None = Field(default=None, description="UUID of the pipeline version used")
+    client_source_path: str | None = Field(default=None, description="Client path that initiated the query")
+
+    # -- User & auth --
+    user: dict[str, Any] | None = Field(
+        default=None, description="User who ran the search (contains 'id', 'given_name', 'family_name')"
+    )
+    api_key: dict[str, Any] | None = Field(default=None, description="API key used (contains 'id', 'name')")
+
+    # -- Feedback, labels, notes --
+    feedback: list[dict[str, Any]] | None = Field(default=None, description="User feedback on this search")
+    labels: list[str] = Field(default_factory=list, description="Labels assigned to this search history record")
+    note: str | None = Field(default=None, description="Free-text note attached to this search history record")
+
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize(cls, data: Any) -> Any:
+        """Extract convenience fields from nested API response structure."""
+        if not isinstance(data, dict):
+            return data
+        # Populate `query` from request.query if not already set at top level
+        if not data.get("query") and isinstance(data.get("request"), dict):
+            data["query"] = data["request"].get("query")
+        # Populate `created_at` from `time` if not already set
+        if not data.get("created_at") and data.get("time"):
+            data["created_at"] = data["time"]
+        return data
 
     @field_validator("feedback", mode="before")
     @classmethod
@@ -33,3 +88,74 @@ def _feedback_to_list(cls, v: Any) -> list[dict[str, Any]] | None:
         if isinstance(v, dict):
             return [v]
         return None
+
+
+class HaystackTraceSpan(BaseModel):
+    """A single span in a Haystack pipeline trace."""
+
+    model_config = {"extra": "allow"}
+
+    span_id: str = Field(description="Unique identifier for this span")
+    parent_span_id: str | None = Field(default=None, description="Parent span ID, if this is a child span")
+    operation_name: str = Field(description="Name of the operation performed in this span")
+    component: str | None = Field(default=None, description="Haystack component that produced this span")
+    start_time: str = Field(description="ISO-8601 timestamp when the span started")
+    end_time: str | None = Field(default=None, description="ISO-8601 timestamp when the span ended")
+    duration_ms: float | None = Field(default=None, description="Duration of the span in milliseconds")
+    tags: dict[str, Any] = Field(default_factory=dict, description="Arbitrary key-value tags attached to the span")
+
+
+class HaystackTraceLog(BaseModel):
+    """A log entry emitted during a Haystack pipeline run."""
+
+    model_config = {"extra": "allow"}
+
+    logger: str = Field(description="Logger name that emitted this entry")
+    level: str = Field(description="Log level (e.g. INFO, WARNING, ERROR)")
+    message: str = Field(description="Log message text")
+    timestamp: str = Field(description="ISO-8601 timestamp of the log entry")
+    extra_fields: dict[str, Any] = Field(default_factory=dict, description="Additional structured fields")
+
+
+class HaystackTraceFailure(BaseModel):
+    """Failure information for a Haystack pipeline run."""
+
+    type: str = Field(description="Exception type name")
+    message: str = Field(description="Exception message")
+    stacktrace: list[str] = Field(description="Stack trace lines")
+
+
+class HaystackTraceV1(BaseModel):
+    """Full Haystack pipeline run trace (schema version haystack-trace/v1)."""
+
+    model_config = {"extra": "allow"}
+
+    schema_version: str = Field(description="Trace schema version identifier")
+    run_id: str = Field(description="Unique run identifier")
+    started_at: str = Field(description="ISO-8601 timestamp when the run started")
+    finished_at: str | None = Field(default=None, description="ISO-8601 timestamp when the run finished")
+    duration_ms: float | None = Field(default=None, description="Total run duration in milliseconds")
+    status: str | None = Field(default=None, description="Run status: 'success' or 'failed'")
+    traces: list[HaystackTraceSpan] = Field(default_factory=list, description="Ordered list of trace spans")
+    logs: list[HaystackTraceLog] = Field(default_factory=list, description="Log entries emitted during the run")
+    failure: HaystackTraceFailure | None = Field(default=None, description="Failure details if the run failed")
+
+
+class PipelineTraceEntry(BaseModel):
+    """A single pipeline trace response from the deepset platform (v2 API format)."""
+
+    model_config = {"extra": "allow"}
+
+    query_id: str = Field(description="Unique identifier for the search query")
+    query: str = Field(description="The search query text that was executed")
+    status: str | None = Field(default=None, description="Run status: 'success' or 'failed'")
+    duration_s: float = Field(description="End-to-end query duration in seconds")
+    created_at: str = Field(description="ISO-8601 timestamp when the query was executed")
+    client_source_path: str | None = Field(default=None, description="Client path that initiated the query")
+    pipeline_version_id: str | None = Field(default=None, description="UUID of the pipeline version used")
+    pipeline_version_name: str | None = Field(default=None, description="Name of the pipeline version used")
+    pipeline_version_number: int | None = Field(default=None, description="Number of the pipeline version used")
+    api_key: dict[str, Any] | None = Field(default=None, description="API key metadata (id, name)")
+    feedback: dict[str, Any] | None = Field(default=None, description="Aggregated user feedback for this query")
+    user_id: str | None = Field(default=None, description="UUID of the user who ran the query")
+    haystack_trace: HaystackTraceV1 | None = Field(default=None, description="Full Haystack pipeline run trace")

src/deepset_mcp/api/search_history/protocols.py

@@ -4,36 +4,85 @@
 
 """Protocols for search history resources."""
 
-from typing import Protocol
+from typing import Literal, Protocol
 
-from deepset_mcp.api.search_history.models import SearchHistoryEntry
+from deepset_mcp.api.search_history.models import PipelineTraceEntry, SearchHistoryEntry
 from deepset_mcp.api.shared_models import PaginatedResponse
 
 
 class SearchHistoryResourceProtocol(Protocol):
     """Protocol defining the interface for search history resources."""
 
     async def list(
-        self, limit: int = 10, after: str | None = None, query_filter: str | None = None
+        self,
+        limit: int = 10,
+        after: str | None = None,
+        query_filter: str | None = None,
+        sort_field: Literal["created_at", "query", "duration", "feedbacks/score"] = "created_at",
+        sort_order: Literal["ASC", "DESC"] = "DESC",
     ) -> PaginatedResponse[SearchHistoryEntry]:
         """List search history entries in the workspace.
 
         :param limit: Maximum number of entries to return per page.
         :param after: Cursor to fetch the next page of results.
         :param query_filter: OData filter expression to narrow results.
+        :param sort_field: Field to sort by.
+        :param sort_order: Sort direction (ASC or DESC).
         :returns: Paginated response of search history entries.
         """
         ...
 
     async def list_pipeline(
-        self, pipeline_name: str, limit: int = 10, after: str | None = None, query_filter: str | None = None
+        self,
+        pipeline_name: str,
+        limit: int = 10,
+        after: str | None = None,
+        query_filter: str | None = None,
+        sort_field: Literal["created_at", "query", "duration", "feedbacks/score"] = "created_at",
+        sort_order: Literal["ASC", "DESC"] = "DESC",
     ) -> PaginatedResponse[SearchHistoryEntry]:
         """List search history entries for a specific pipeline with pagination.
 
         :param pipeline_name: Name of the pipeline.
         :param limit: Maximum number of entries to return per page.
         :param after: Cursor to fetch the next page of results.
         :param query_filter: OData filter expression to narrow results.
+        :param sort_field: Field to sort by.
+        :param sort_order: Sort direction (ASC or DESC).
         :returns: Paginated response of search history entries (most recent first).
         """
         ...
+
+    async def list_pipeline_traces(
+        self,
+        pipeline_name: str,
+        limit: int = 10,
+        after: str | None = None,
+        query_filter: str | None = None,
+        sort_field: Literal["created_at", "query", "duration", "feedbacks/score"] = "created_at",
+        sort_order: Literal["ASC", "DESC"] = "DESC",
+    ) -> PaginatedResponse[PipelineTraceEntry]:
+        """List Haystack pipeline run traces with pagination, filtering, and sorting.
+
+        :param pipeline_name: Name of the pipeline.
+        :param limit: Maximum number of trace entries to return per page.
+        :param after: Cursor to fetch the next page of results.
+        :param query_filter: OData filter expression to narrow results.
+        :param sort_field: Field to sort by.
+        :param sort_order: Sort direction (ASC or DESC).
+        :returns: Paginated response of pipeline trace entries.
+        """
+        ...
+
+    async def get_pipeline_trace(
+        self,
+        pipeline_name: str,
+        query_id: str,
+    ) -> PipelineTraceEntry | None:
+        """Get the Haystack pipeline run trace for a single search history record.
+
+        :param pipeline_name: Name of the pipeline.
+        :param query_id: UUID of the search history query.
+        :returns: The trace entry, or None if not found.
+        """
+        ...