Integrasi Model Context Protocol (MCP) pada backend Python rentan terhadap degradasi performa akibat context window bloat. Masalah ini terjadi ketika server meneruskan data riwayat eksekusi tool atau resource ke model tanpa pembersihan, memicu lonjakan konsumsi memori (OOM) dan latensi inferensi LLM yang tinggi.

Gejala: Memory Spike dan Latensi LLM Membengkak

Saat menjalankan MCP server lokal menggunakan stdio atau SSE (Server-Sent Events), dua gejala utama yang sering muncul adalah:

  • Degradasi Latensi Ekstrem: Waktu respons model meningkat dari <2 detik menjadi >30 detik per interaksi karena LLM harus memproses token input yang terus membengkak.
  • Memory Spike / OOM: Proses Python lokal mengalami peningkatan alokasi RAM bertahap hingga terminated oleh OS (Out Of Memory Killer), terutama saat menangani output query database atau log files berukuran besar.

Root Cause Analysis

Investigasi log JSON-RPC menunjukkan dua penyebab utama:

  1. Akumulasi State Tanpa Batas: Server menyimpan riwayat panggilan tool sebelumnya di dalam memori sesi tanpa mekanisme eviction.
  2. Payload Mentah Tanpa Pruning: Output fungsi (misalnya fetch logs atau data tabel) dikembalikan utuh ke client JSON-RPC tanpa proses filtering field yang tidak relevan, menyita ribuan token yang tidak esensial bagi konteks model.

Solusi: Sliding-Window Buffer dan Context Cleaner

Pola perbaikan yang efisien mengombinasikan generator pembersih data terstruktur dan buffer berbasis kuota token untuk menjaga payload JSON-RPC tetap berada di bawah ambang batas aman.

import json
from typing import List, Dict, Any, Generator

def sanitize_record(record: Dict[str, Any], allowed_keys: set) -> Dict[str, Any]:
    """Hapus keys payload yang tidak esensial untuk menghemat token."""
    return {k: v for k, v in record.items() if k in allowed_keys}

def estimate_tokens(text: str) -> int:
    """Estimasi token konservatif berbasis karakter standar (1 token ~= 4 karakter)."""
    # ponytail: gunakan tiktoken jika membutuhkan akurasi token spesifik model
    return max(1, len(text) // 4)

def clean_context_stream(records: List[Dict[str, Any]], allowed_keys: set) -> Generator[Dict[str, Any], None, None]:
    """Generator pembersih record data satu per satu guna meminimalkan alokasi heap."""
    for item in records:
        yield sanitize_record(item, allowed_keys)

class TokenSlidingWindow:
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
        self.buffer: List[Dict[str, Any]] = []
        self.current_tokens = 0

    def append(self, message: Dict[str, Any]) -> None:
        payload_str = json.dumps(message)
        tokens = estimate_tokens(payload_str)

        while self.buffer and (self.current_tokens + tokens > self.max_tokens):
            dropped = self.buffer.pop(0)
            self.current_tokens -= estimate_tokens(json.dumps(dropped))

        self.buffer.append(message)
        self.current_tokens += tokens

    def get_messages(self) -> List[Dict[str, Any]]:
        return list(self.buffer)

Pendekatan di atas memangkas penggunaan memori melalui generator dan membatasi ukuran konteks yang dikirim kembali ke client MCP, mencegah payload JSON-RPC meluap.

Unit Test: Verifikasi Limit Token Context

Uji batas token berikut memastikan mekanisme sliding window menghapus item lama saat limit terlampaui:

def test_sliding_window_token_limit():
    # Setup window kecil untuk testing (maksimal 20 token)
    window = TokenSlidingWindow(max_tokens=20)
    
    msg_1 = {"role": "tool", "content": "data 1 yang panjang"}
    msg_2 = {"role": "tool", "content": "data 2 yang panjang"}
    msg_3 = {"role": "tool", "content": "data 3 yang panjang"}
    
    window.append(msg_1)
    window.append(msg_2)
    assert len(window.get_messages()) >= 1
    
    # Penambahan msg_3 harus menyingkirkan msg_1 jika kuota terlampaui
    window.append(msg_3)
    messages = window.get_messages()
    
    assert msg_1 not in messages, "Record lama (msg_1) harus di-evict."
    assert messages[-1] == msg_3, "Record terbaru (msg_3) harus ada di buffer."
    assert window.current_tokens <= window.max_tokens, "Total token melebihi limit."

if __name__ == "__main__":
    test_sliding_window_token_limit()
    print("Semua verifikasi limit context lolos.")

Trade-off dan Limitasi

Membatasi konteks dengan sliding window memiliki konsekuensi logis: model kehilangan referensi ke instruksi atau output tool paling awal. Untuk mitigasi lanjutan pada MCP server produksi:

  • Gunakan teknik summarization pass jika interaksi masa lalu tetap harus diingat secara garis besar.
  • Terapkan pagination langsung pada level MCP tool ketimbang mengembalikan data mentah dalam jumlah ribuan baris sekaligus.