Skip to content

base

ipw.agents.mcp.base

Base MCP server with telemetry integration.

All MCP servers (local models, cloud APIs, tools) inherit from BaseMCPServer, which automatically captures energy, power, cost, and latency metrics around every execution.

MCPToolResult dataclass

Result from MCP tool execution with telemetry.

Source code in intelligence-per-watt/src/ipw/agents/mcp/base.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass
class MCPToolResult:
    """Result from MCP tool execution with telemetry."""

    content: str
    """Response text from tool/model"""

    usage: Dict[str, int] = field(default_factory=dict)
    """Token counts: prompt_tokens, completion_tokens, total_tokens"""

    cost_usd: Optional[float] = None
    """API cost in USD (for cloud APIs)"""

    telemetry_samples: List[Any] = field(default_factory=list)
    """Energy/power/memory readings during execution"""

    latency_seconds: float = 0.0
    """Wall-clock execution time"""

    ttft_seconds: Optional[float] = None
    """Time to first token (for streaming APIs)"""

    metadata: Dict[str, Any] = field(default_factory=dict)
    """Additional tool-specific metadata"""

content instance-attribute

Response text from tool/model

usage = field(default_factory=dict) class-attribute instance-attribute

Token counts: prompt_tokens, completion_tokens, total_tokens

cost_usd = None class-attribute instance-attribute

API cost in USD (for cloud APIs)

telemetry_samples = field(default_factory=list) class-attribute instance-attribute

Energy/power/memory readings during execution

latency_seconds = 0.0 class-attribute instance-attribute

Wall-clock execution time

ttft_seconds = None class-attribute instance-attribute

Time to first token (for streaming APIs)

metadata = field(default_factory=dict) class-attribute instance-attribute

Additional tool-specific metadata

BaseMCPServer

Bases: ABC

Base class for all MCP servers with automatic telemetry capture.

All subclasses must implement _execute_impl() which performs the actual tool invocation. The base class wraps this with telemetry collection.

Example

class MyTool(BaseMCPServer): def _execute_impl(self, prompt: str, **params) -> MCPToolResult: response = self.api.call(prompt) return MCPToolResult( content=response.text, usage={"prompt_tokens": 100, "completion_tokens": 50}, cost_usd=0.001 )

Source code in intelligence-per-watt/src/ipw/agents/mcp/base.py 
 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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class BaseMCPServer(ABC):
    """Base class for all MCP servers with automatic telemetry capture.

    All subclasses must implement _execute_impl() which performs the actual
    tool invocation. The base class wraps this with telemetry collection.

    Example:
        class MyTool(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                response = self.api.call(prompt)
                return MCPToolResult(
                    content=response.text,
                    usage={"prompt_tokens": 100, "completion_tokens": 50},
                    cost_usd=0.001
                )
    """

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

        Args:
            name: Tool name for logging/tracking
            telemetry_collector: Energy monitor collector. If None, runs without telemetry.
            event_recorder: EventRecorder for per-action tracking. If None, no events recorded.
        """
        self.name = name
        self.telemetry_collector = telemetry_collector
        self.event_recorder = event_recorder

    def execute(self, prompt: str, **params: Any) -> MCPToolResult:
        """Execute tool with automatic telemetry capture.

        Args:
            prompt: Input prompt/query for tool
            **params: Additional tool-specific parameters

        Returns:
            MCPToolResult with content, usage, cost, and telemetry samples
        """
        start_time = time.time()

        # Get model info for event recording (subclasses can override these)
        model_id = getattr(self, "model_path", None) or getattr(self, "model_name", self.name)
        model_alias = getattr(self, "model_name", self.name)
        backend = self._get_backend()

        # Record start event
        if self.event_recorder is not None:
            self.event_recorder.record(
                "submodel_call_start",
                model_id=model_id,
                model_alias=model_alias,
                backend=backend,
                tool_name=self.name,
            )

        # Execute with telemetry if available
        if HAS_TELEMETRY and self.telemetry_collector is not None and TelemetrySession is not None:
            with TelemetrySession(self.telemetry_collector) as session:
                result = self._execute_impl(prompt, **params)
                end_time = time.time()
                result.telemetry_samples = list(session.window(start_time, end_time))
                result.latency_seconds = end_time - start_time
        else:
            # Execute without telemetry
            result = self._execute_impl(prompt, **params)
            end_time = time.time()
            result.latency_seconds = end_time - start_time

        # Record end event
        if self.event_recorder is not None:
            self.event_recorder.record(
                "submodel_call_end",
                model_id=model_id,
                model_alias=model_alias,
                backend=backend,
                tool_name=self.name,
                total_tokens=result.usage.get("total_tokens", 0),
                prompt_tokens=result.usage.get("prompt_tokens", 0),
                completion_tokens=result.usage.get("completion_tokens", 0),
                cost_usd=result.cost_usd,
                latency_seconds=result.latency_seconds,
            )

        return result

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

        Returns:
            Backend identifier (e.g., 'vllm', 'ollama', 'openai')
        """
        # Default: extract from name if it contains a prefix like "vllm:" or "ollama:"
        if ":" in self.name:
            return self.name.split(":")[0]
        return "unknown"

    @abstractmethod
    def _execute_impl(self, prompt: str, **params: Any) -> MCPToolResult:
        """Implement tool execution logic.

        Subclasses must override this to provide actual tool functionality.

        Args:
            prompt: Input prompt/query
            **params: Tool-specific parameters

        Returns:
            MCPToolResult (telemetry_samples and latency_seconds will be added by base class)
        """
        raise NotImplementedError

    def health_check(self) -> bool:
        """Check if tool is available and healthy.

        Returns:
            True if tool is operational, False otherwise
        """
        try:
            result = self.execute("test", timeout=5)
            return result.content is not None
        except Exception:
            return False

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name!r})"

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

Initialize MCP server.

Parameters:

Name Type Description Default
name str

Tool name for logging/tracking

 required
telemetry_collector Optional[Any]

Energy monitor collector. If None, runs without telemetry.

 None
event_recorder Optional[Any]

EventRecorder for per-action tracking. If None, no events recorded.

 None
Source code in intelligence-per-watt/src/ipw/agents/mcp/base.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    name: str,
    telemetry_collector: Optional[Any] = None,
    event_recorder: Optional[Any] = None,
):
    """Initialize MCP server.

    Args:
        name: Tool name for logging/tracking
        telemetry_collector: Energy monitor collector. If None, runs without telemetry.
        event_recorder: EventRecorder for per-action tracking. If None, no events recorded.
    """
    self.name = name
    self.telemetry_collector = telemetry_collector
    self.event_recorder = event_recorder

execute(prompt, **params)

Execute tool with automatic telemetry capture.

Parameters:

Name Type Description Default
prompt str

Input prompt/query for tool

 required
**params Any

Additional tool-specific parameters

 {}

Returns:

Type Description
MCPToolResult

MCPToolResult with content, usage, cost, and telemetry samples
Source code in intelligence-per-watt/src/ipw/agents/mcp/base.py 
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
def execute(self, prompt: str, **params: Any) -> MCPToolResult:
    """Execute tool with automatic telemetry capture.

    Args:
        prompt: Input prompt/query for tool
        **params: Additional tool-specific parameters

    Returns:
        MCPToolResult with content, usage, cost, and telemetry samples
    """
    start_time = time.time()

    # Get model info for event recording (subclasses can override these)
    model_id = getattr(self, "model_path", None) or getattr(self, "model_name", self.name)
    model_alias = getattr(self, "model_name", self.name)
    backend = self._get_backend()

    # Record start event
    if self.event_recorder is not None:
        self.event_recorder.record(
            "submodel_call_start",
            model_id=model_id,
            model_alias=model_alias,
            backend=backend,
            tool_name=self.name,
        )

    # Execute with telemetry if available
    if HAS_TELEMETRY and self.telemetry_collector is not None and TelemetrySession is not None:
        with TelemetrySession(self.telemetry_collector) as session:
            result = self._execute_impl(prompt, **params)
            end_time = time.time()
            result.telemetry_samples = list(session.window(start_time, end_time))
            result.latency_seconds = end_time - start_time
    else:
        # Execute without telemetry
        result = self._execute_impl(prompt, **params)
        end_time = time.time()
        result.latency_seconds = end_time - start_time

    # Record end event
    if self.event_recorder is not None:
        self.event_recorder.record(
            "submodel_call_end",
            model_id=model_id,
            model_alias=model_alias,
            backend=backend,
            tool_name=self.name,
            total_tokens=result.usage.get("total_tokens", 0),
            prompt_tokens=result.usage.get("prompt_tokens", 0),
            completion_tokens=result.usage.get("completion_tokens", 0),
            cost_usd=result.cost_usd,
            latency_seconds=result.latency_seconds,
        )

    return result

health_check()

Check if tool is available and healthy.

Returns:

Type Description
bool

True if tool is operational, False otherwise
Source code in intelligence-per-watt/src/ipw/agents/mcp/base.py 
185
186
187
188
189
190
191
192
193
194
195
def health_check(self) -> bool:
    """Check if tool is available and healthy.

    Returns:
        True if tool is operational, False otherwise
    """
    try:
        result = self.execute("test", timeout=5)
        return result.content is not None
    except Exception:
        return False