Skip to content

base

ipw.agents.mcp.retrieval.base

Base classes for retrieval MCP servers.

All retrieval servers inherit from BaseRetrievalServer, which extends BaseMCPServer to provide automatic telemetry capture for energy, power, and latency.

Document dataclass

A document for retrieval indexing.

Attributes:

Name Type Description
id str

Unique document identifier
content str

Document text content
metadata Dict[str, Any]

Optional metadata (e.g., title, source, date)
Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass
class Document:
    """A document for retrieval indexing.

    Attributes:
        id: Unique document identifier
        content: Document text content
        metadata: Optional metadata (e.g., title, source, date)
    """

    id: str
    content: str
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self):
        """Validate document."""
        if not self.id:
            raise ValueError("Document id cannot be empty")
        if not self.content:
            raise ValueError("Document content cannot be empty")

__post_init__()

Validate document.

Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
30
31
32
33
34
35
def __post_init__(self):
    """Validate document."""
    if not self.id:
        raise ValueError("Document id cannot be empty")
    if not self.content:
        raise ValueError("Document content cannot be empty")

RetrievalResult dataclass

A single retrieval result with score.

Attributes:

Name Type Description
document Document

The retrieved document
score float

Relevance score (higher is better)
highlights List[str]

Optional highlighted text snippets
Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@dataclass
class RetrievalResult:
    """A single retrieval result with score.

    Attributes:
        document: The retrieved document
        score: Relevance score (higher is better)
        highlights: Optional highlighted text snippets
    """

    document: Document
    score: float
    highlights: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "id": self.document.id,
            "content": self.document.content,
            "metadata": self.document.metadata,
            "score": self.score,
            "highlights": self.highlights,
        }

to_dict()

Convert to dictionary for serialization.

Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
52
53
54
55
56
57
58
59
60
def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary for serialization."""
    return {
        "id": self.document.id,
        "content": self.document.content,
        "metadata": self.document.metadata,
        "score": self.score,
        "highlights": self.highlights,
    }

BaseRetrievalServer

Bases: BaseMCPServer

Base class for retrieval servers with automatic telemetry.

Subclasses must implement: - _execute_impl(): Perform the actual retrieval - index_documents(): Index a list of documents - clear_index(): Clear all indexed documents

Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class BaseRetrievalServer(BaseMCPServer):
    """Base class for retrieval servers with automatic telemetry.

    Subclasses must implement:
    - _execute_impl(): Perform the actual retrieval
    - index_documents(): Index a list of documents
    - clear_index(): Clear all indexed documents
    """

    def __init__(
        self,
        name: str,
        telemetry_collector: Optional[Any] = None,
        event_recorder: Optional[Any] = None,
    ):
        """Initialize retrieval server.

        Args:
            name: Server name (e.g., "retrieval:bm25")
            telemetry_collector: Energy monitor collector for telemetry
            event_recorder: EventRecorder for per-action tracking
        """
        super().__init__(
            name=name,
            telemetry_collector=telemetry_collector,
            event_recorder=event_recorder,
        )
        self._document_count = 0

    def _get_backend(self) -> str:
        """Get the backend type for this server."""
        return "retrieval"

    @abstractmethod
    def index_documents(self, documents: List[Document]) -> int:
        """Index a list of documents.

        Args:
            documents: List of documents to index

        Returns:
            Number of documents successfully indexed
        """
        raise NotImplementedError

    @abstractmethod
    def clear_index(self) -> None:
        """Clear all indexed documents."""
        raise NotImplementedError

    @property
    def document_count(self) -> int:
        """Get the number of indexed documents."""
        return self._document_count

    def _format_results(
        self,
        results: List[RetrievalResult],
        include_scores: bool = True,
        include_metadata: bool = False,
    ) -> str:
        """Format retrieval results as text."""
        if not results:
            return "No results found."

        lines = []
        for i, result in enumerate(results, 1):
            header = f"[{i}] {result.document.id}"
            if include_scores:
                header += f" (score: {result.score:.4f})"
            lines.append(header)

            if include_metadata and result.document.metadata:
                meta_str = ", ".join(
                    f"{k}={v}" for k, v in result.document.metadata.items()
                )
                lines.append(f"  Metadata: {meta_str}")

            if result.highlights:
                for highlight in result.highlights:
                    lines.append(f"  ...{highlight}...")
            else:
                content = result.document.content
                if len(content) > 200:
                    content = content[:200] + "..."
                lines.append(f"  {content}")

            lines.append("")

        return "\n".join(lines)

    def health_check(self) -> bool:
        """Check if retrieval server is operational."""
        try:
            return self._document_count >= 0
        except Exception:
            return False

document_count property

Get the number of indexed documents.

__init__(name, telemetry_collector=None, event_recorder=None)

Initialize retrieval server.

Parameters:

Name Type Description Default
name str

Server name (e.g., "retrieval:bm25")

 required
telemetry_collector Optional[Any]

Energy monitor collector for telemetry

 None
event_recorder Optional[Any]

EventRecorder for per-action tracking

 None
Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    name: str,
    telemetry_collector: Optional[Any] = None,
    event_recorder: Optional[Any] = None,
):
    """Initialize retrieval server.

    Args:
        name: Server name (e.g., "retrieval:bm25")
        telemetry_collector: Energy monitor collector for telemetry
        event_recorder: EventRecorder for per-action tracking
    """
    super().__init__(
        name=name,
        telemetry_collector=telemetry_collector,
        event_recorder=event_recorder,
    )
    self._document_count = 0

index_documents(documents) abstractmethod

Index a list of documents.

Parameters:

Name Type Description Default
documents List[Document]

List of documents to index

 required

Returns:

Type Description
int

Number of documents successfully indexed
Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
@abstractmethod
def index_documents(self, documents: List[Document]) -> int:
    """Index a list of documents.

    Args:
        documents: List of documents to index

    Returns:
        Number of documents successfully indexed
    """
    raise NotImplementedError

clear_index() abstractmethod

Clear all indexed documents.

Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
108
109
110
111
@abstractmethod
def clear_index(self) -> None:
    """Clear all indexed documents."""
    raise NotImplementedError

health_check()

Check if retrieval server is operational.

Source code in intelligence-per-watt/src/ipw/agents/mcp/retrieval/base.py 
154
155
156
157
158
159
def health_check(self) -> bool:
    """Check if retrieval server is operational."""
    try:
        return self._document_count >= 0
    except Exception:
        return False