Skip to content

anthropic_server

ipw.agents.mcp.anthropic_server

Anthropic MCP server with cost tracking.

AnthropicMCPServer

Bases: BaseMCPServer

MCP server for Anthropic Claude models with cost tracking.

Example

server = AnthropicMCPServer( model_name="claude-sonnet-4-5-20250929", api_key=os.getenv("ANTHROPIC_API_KEY") )

result = server.execute("Write a haiku about AI") print(result.content) print(f"Cost: ${result.cost_usd:.4f}")

Source code in intelligence-per-watt/src/ipw/agents/mcp/anthropic_server.py 
 12
 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
 46
 47
 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
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
class AnthropicMCPServer(BaseMCPServer):
    """MCP server for Anthropic Claude models with cost tracking.

    Example:
        server = AnthropicMCPServer(
            model_name="claude-sonnet-4-5-20250929",
            api_key=os.getenv("ANTHROPIC_API_KEY")
        )

        result = server.execute("Write a haiku about AI")
        print(result.content)
        print(f"Cost: ${result.cost_usd:.4f}")
    """

    def __init__(
        self,
        model_name: str,
        api_key: Optional[str] = None,
        telemetry_collector: Optional[Any] = None,
        **anthropic_params: Any,
    ):
        """Initialize Anthropic MCP server.

        Args:
            model_name: Claude model name
            api_key: Anthropic API key (or set ANTHROPIC_API_KEY env var)
            telemetry_collector: Energy monitor collector
            **anthropic_params: Additional params (temperature, max_tokens, etc.)
        """
        super().__init__(
            name=f"anthropic:{model_name}",
            telemetry_collector=telemetry_collector,
        )

        self.model_name = model_name
        self.anthropic_params = anthropic_params

        # Lazy import: anthropic is optional
        try:
            from anthropic import Anthropic
        except ImportError:
            raise ImportError(
                "anthropic package is required for AnthropicMCPServer. "
                "Install with: pip install anthropic"
            )

        # Initialize Anthropic client
        self._client = Anthropic(api_key=api_key)

        # Get pricing for this model
        self.pricing = ANTHROPIC_PRICING.get(model_name)
        if not self.pricing:
            print(f"Warning: No pricing info for {model_name}, using Sonnet 4.5 rates")
            self.pricing = ANTHROPIC_PRICING["claude-sonnet-4-5-20250929"]

    def _execute_impl(self, prompt: str, **params: Any) -> MCPToolResult:
        """Execute Anthropic API call with cost tracking."""
        from anthropic import AnthropicError

        # Merge default params with request params
        payload = {**self.anthropic_params, **params}
        payload["model"] = self.model_name
        payload["messages"] = [{"role": "user", "content": prompt}]
        payload["max_tokens"] = payload.get("max_tokens", 4096)
        payload["stream"] = True

        # Call Anthropic API
        start = time.perf_counter()
        try:
            stream = self._client.messages.create(**payload)
        except AnthropicError as exc:
            raise RuntimeError(f"Anthropic error for {self.model_name}: {exc}") from exc

        # Consume stream and collect response
        content_chunks: list[str] = []
        ttft_ms: Optional[float] = None
        input_tokens = 0
        output_tokens = 0

        with stream as event_stream:
            for event in event_stream:
                if event.type == "content_block_delta":
                    if hasattr(event.delta, "text"):
                        if ttft_ms is None:
                            ttft_ms = (time.perf_counter() - start) * 1000
                        content_chunks.append(event.delta.text)

                elif event.type == "message_start":
                    if hasattr(event.message, "usage"):
                        input_tokens = event.message.usage.input_tokens

                elif event.type == "message_delta":
                    if hasattr(event, "usage"):
                        output_tokens = event.usage.output_tokens

        content = "".join(content_chunks)

        # Calculate cost
        cost_usd = calculate_cost("anthropic", self.model_name, input_tokens, output_tokens)

        return MCPToolResult(
            content=content,
            usage={
                "prompt_tokens": input_tokens,
                "completion_tokens": output_tokens,
                "total_tokens": input_tokens + output_tokens,
            },
            cost_usd=cost_usd,
            ttft_seconds=(ttft_ms / 1000.0) if ttft_ms else None,
            metadata={
                "model": self.model_name,
                "backend": "anthropic",
                "pricing_input_per_1m": self.pricing["input"],
                "pricing_output_per_1m": self.pricing["output"],
            },
        )

    def health_check(self) -> bool:
        """Check if Anthropic API is accessible."""
        try:
            response = self._client.messages.create(
                model=self.model_name,
                messages=[{"role": "user", "content": "test"}],
                max_tokens=1,
            )
            return response is not None
        except Exception:
            return False

__init__(model_name, api_key=None, telemetry_collector=None, **anthropic_params)

Initialize Anthropic MCP server.

Parameters:

Name Type Description Default
model_name str

Claude model name

 required
api_key Optional[str]

Anthropic API key (or set ANTHROPIC_API_KEY env var)

 None
telemetry_collector Optional[Any]

Energy monitor collector

 None
**anthropic_params Any

Additional params (temperature, max_tokens, etc.)

 {}
Source code in intelligence-per-watt/src/ipw/agents/mcp/anthropic_server.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    model_name: str,
    api_key: Optional[str] = None,
    telemetry_collector: Optional[Any] = None,
    **anthropic_params: Any,
):
    """Initialize Anthropic MCP server.

    Args:
        model_name: Claude model name
        api_key: Anthropic API key (or set ANTHROPIC_API_KEY env var)
        telemetry_collector: Energy monitor collector
        **anthropic_params: Additional params (temperature, max_tokens, etc.)
    """
    super().__init__(
        name=f"anthropic:{model_name}",
        telemetry_collector=telemetry_collector,
    )

    self.model_name = model_name
    self.anthropic_params = anthropic_params

    # Lazy import: anthropic is optional
    try:
        from anthropic import Anthropic
    except ImportError:
        raise ImportError(
            "anthropic package is required for AnthropicMCPServer. "
            "Install with: pip install anthropic"
        )

    # Initialize Anthropic client
    self._client = Anthropic(api_key=api_key)

    # Get pricing for this model
    self.pricing = ANTHROPIC_PRICING.get(model_name)
    if not self.pricing:
        print(f"Warning: No pricing info for {model_name}, using Sonnet 4.5 rates")
        self.pricing = ANTHROPIC_PRICING["claude-sonnet-4-5-20250929"]

health_check()

Check if Anthropic API is accessible.

Source code in intelligence-per-watt/src/ipw/agents/mcp/anthropic_server.py 
129
130
131
132
133
134
135
136
137
138
139
def health_check(self) -> bool:
    """Check if Anthropic API is accessible."""
    try:
        response = self._client.messages.create(
            model=self.model_name,
            messages=[{"role": "user", "content": "test"}],
            max_tokens=1,
        )
        return response is not None
    except Exception:
        return False