MCP Context Caching and Prompt Compression

┌────────────────────────────────────────────────────────┐
│ Turn 1: 25,000 tokens (Tool Schemas) + 200 user tokens │
├────────────────────────────────────────────────────────┤
│ Turn 2: 25,000 tokens (Tool Schemas) + 1,200 tokens    │
├────────────────────────────────────────────────────────┤
│ ...                                                    │
├────────────────────────────────────────────────────────┤
│ Turn 20: 25,000 tokens + 15,000 history tokens         │
└────────────────────────────────────────────────────────┘
Cumulative Uncached Cost: ~650,000 input tokens per session!

# cached_mcp_client.py
import anthropic
from typing import List, Dict, Any

def prepare_cached_mcp_tools(mcp_tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Format MCP tools for Anthropic API and place an ephemeral cache breakpoint
    at the end of the tool catalog.
    """
    formatted_tools = []
    
    for i, tool in enumerate(mcp_tools):
        tool_def = {
            "name": tool["name"],
            "description": tool.get("description", ""),
            "input_schema": tool.get("inputSchema", {})
        }
        
        # Apply cache breakpoint to the final tool definition
        if i == len(mcp_tools) - 1:
            tool_def["cache_control"] = {"type": "ephemeral"}
            
        formatted_tools.append(tool_def)
        
    return formatted_tools

client = anthropic.Anthropic()

def execute_agent_turn(messages: list, mcp_tools: list):
    cached_tools = prepare_cached_mcp_tools(mcp_tools)
    
    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=2048,
        tools=cached_tools,
        messages=messages
    )
    
    # Inspect cache read hits
    usage = response.usage
    print(f"Tokens written to cache: {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"Tokens read from cache: {getattr(usage, 'cache_read_input_tokens', 0)}")
    return response

// schema-compressor.ts
export function compressMcpToolSchema(rawTool: any): any {
  const schema = { ...rawTool.inputSchema };

  // Recursive stripper for JSON Schema bloat
  function stripBloat(obj: any): any {
    if (!obj || typeof obj !== "object") return obj;
    if (Array.isArray(obj)) return obj.map(stripBloat);

    const cleaned: any = {};
    for (const [key, value] of Object.entries(obj)) {
      // Discard cosmetic schema keys not needed for LLM argument parsing
      if (["title", "$schema", "examples", "default"].includes(key)) {
        continue;
      }
      cleaned[key] = stripBloat(value);
    }
    return cleaned;
  }

  return {
    name: rawTool.name,
    description: rawTool.description?.substring(0, 160) || "",
    input_schema: stripBloat(schema)
  };
}