Skip to content

test_base_mcp

ipw.tests.agents.mcp.test_base_mcp

Tests for agents/mcp/base.py — MCPToolResult and BaseMCPServer.

TestMCPToolResult

Test MCPToolResult dataclass fields.

Source code in intelligence-per-watt/src/ipw/tests/agents/mcp/test_base_mcp.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class TestMCPToolResult:
    """Test MCPToolResult dataclass fields."""

    def test_default_values(self) -> None:
        result = MCPToolResult(content="hello")
        assert result.content == "hello"
        assert result.usage == {}
        assert result.cost_usd is None
        assert result.telemetry_samples == []
        assert result.latency_seconds == 0.0
        assert result.ttft_seconds is None
        assert result.metadata == {}

    def test_all_fields(self) -> None:
        result = MCPToolResult(
            content="answer",
            usage={"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
            cost_usd=0.001,
            telemetry_samples=[{"power": 100}],
            latency_seconds=0.5,
            ttft_seconds=0.1,
            metadata={"model": "test"},
        )
        assert result.cost_usd == 0.001
        assert result.latency_seconds == 0.5
        assert result.ttft_seconds == 0.1
        assert result.usage["total_tokens"] == 15

    def test_usage_dict_is_independent(self) -> None:
        r1 = MCPToolResult(content="a")
        r2 = MCPToolResult(content="b")
        r1.usage["key"] = 1
        assert "key" not in r2.usage

TestBaseMCPServer

Test BaseMCPServer contract.

Source code in intelligence-per-watt/src/ipw/tests/agents/mcp/test_base_mcp.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 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
class TestBaseMCPServer:
    """Test BaseMCPServer contract."""

    def test_cannot_instantiate_directly(self) -> None:
        with pytest.raises(TypeError):
            BaseMCPServer(name="test")

    def test_concrete_subclass(self) -> None:
        class DummyServer(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                return MCPToolResult(content=f"echo: {prompt}")

        server = DummyServer(name="dummy")
        assert server.name == "dummy"
        assert repr(server) == "DummyServer(name='dummy')"

    def test_execute_wraps_impl(self) -> None:
        class DummyServer(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                return MCPToolResult(content=f"echo: {prompt}")

        server = DummyServer(name="dummy")
        result = server.execute("hello")
        assert result.content == "echo: hello"
        assert result.latency_seconds > 0

    def test_execute_records_events(self) -> None:
        recorder = EventRecorder()

        class DummyServer(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                return MCPToolResult(
                    content="ok",
                    usage={"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
                    cost_usd=0.001,
                )

        server = DummyServer(name="dummy", event_recorder=recorder)
        server.execute("test")

        events = recorder.get_events()
        assert len(events) == 2
        assert events[0].event_type == "submodel_call_start"
        assert events[1].event_type == "submodel_call_end"
        assert events[1].metadata["total_tokens"] == 15
        assert events[1].metadata["cost_usd"] == 0.001

    def test_get_backend_from_name(self) -> None:
        class DummyServer(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                return MCPToolResult(content="ok")

        server = DummyServer(name="vllm:model")
        assert server._get_backend() == "vllm"

    def test_get_backend_unknown(self) -> None:
        class DummyServer(BaseMCPServer):
            def _execute_impl(self, prompt: str, **params) -> MCPToolResult:
                return MCPToolResult(content="ok")

        server = DummyServer(name="calculator")
        assert server._get_backend() == "unknown"