Sistem multi-agent berbasis Python sering kali mengalami worker hang ketika mengeksekusi banyak tool call secara bersamaan melalui Model Context Protocol (MCP). Gejala ini biasanya ditandai dengan orchestrator yang berhenti merespons, konsumsi CPU turun ke nol persen, dan task pipeline tertahan tanpa pesan error atau stack trace eksplisit.

Masalah ini berakar pada starvation dan deadlock di dalam single event loop asyncio saat berinteraksi dengan transport STDIO child process MCP. Artikel ini membedah log traces, akar masalah arsitektural, dan implementasi migrasi ke AnyIO thread worker isolation dengan manajemen timeout yang ketat.

Gejala dan Log Traces Worker Hang

Saat pipeline multi-agent memicu multiple sub-agent (misalnya perencana, analis kode, dan eksekutor bash) secara paralel menggunakan asyncio.gather, request RPC terkirim ke child process MCP via pipeline STDIO. Ketika payload bertambah besar atau sub-process menjalankan pemrosesan sinkron, worker berhenti menerima respons.

Pemeriksaan thread dump via sinyal OS (kill -3 <PID> atau modul faulthandler) menunjukkan kondisi berikut:

Thread 0x00007f9c882a1640 (most recent call first):
  File "/usr/lib/python3.11/asyncio/base_events.py", line 1909 in _run_once
    event_list = self._selector.select(timeout)
  File "/usr/lib/python3.11/asyncio/base_events.py", line 607 in run_forever
    self._run_once()
  File "/usr/lib/python3.11/asyncio/runners.py", line 118 in run
    return self._loop.run_until_complete(task)
  File "agent_orchestrator.py", line 84 in execute_parallel_tools
    results = loop.run_until_complete(asyncio.gather(*tasks))
  File "main.py", line 22 in <module>
    execute_parallel_tools(agent_payloads)

Log aplikasi terhenti tepat setelah pengiriman request JSON-RPC, tanpa respon balik dari pipe:

[2024-10-12 10:14:02,112] INFO: orchestrator: Dispatching tool 'read_file' to MCP-Worker-1
[2024-10-12 10:14:02,115] INFO: orchestrator: Dispatching tool 'git_diff' to MCP-Worker-2
[2024-10-12 10:14:02,116] DEBUG: mcp.transport.stdio: Sent JSON-RPC frame ID 109 to Worker-1
[2024-10-12 10:14:02,118] DEBUG: mcp.transport.stdio: Sent JSON-RPC frame ID 110 to Worker-2
# <Hangs indefinitely. Event loop idle, buffers locked>

Akar Masalah: Asyncio Event Loop Starvation dan Deadlock Buffer STDIO

Implementasi transport standar pada MCP sering menggunakan pipe stream OS (stdin, stdout) untuk komunikasi antar process. Terjadinya deadlock disebabkan oleh kombinasi dua faktor utama:

  1. Sinkronisasi Blocking di dalam Callback Async: Jika tool atau adapter MCP menjalankan library sinkron (misalnya requests, subprocess.run, atau akses file disk besar) di dalam context yang sama, event loop thread utama terblokir. Selama thread utama terblokir, callback reader untuk pipe stdout child process tidak dijalankan.
  2. OS Pipe Buffer Saturation: Sistem operasi menetapkan ukuran buffer terbatas untuk pipe (umumnya 64KB pada Linux). Jika child process menulis data melebihi kapasitas buffer sementara parent process event loop-nya terhambat dan belum menguras (drain) pipe, child process akan terblokir pada pemanggilan write(). Kedua process akhirnya saling menunggu (deadlock).

Solusi: Thread Isolation Menggunakan AnyIO dan Timeout Handling

Untuk mengeliminasi risiko blocking pada transport pipe, pemanggilan tool MCP yang bersifat I/O intensif atau berpotensi sinkron harus diisolasi ke worker thread terpisah di luar event loop utama. Library anyio menyediakan abstraksi structured concurrency yang kompatibel dengan asyncio dan trio serta mempermudah eksekusi kode sinkron tanpa mengganggu loop.

Implementasi MCP Tool Runner yang Aman

Berikut adalah implementasi refaktor untuk mengisolasi transport worker MCP dan menambahkan batas eksekusi per tool call:

import sys
import anyio
from typing import Any, Dict
from anyio import to_thread, fail_after

class DeadlockSafeMCPClient:
    def __init__(self, command: list[str]):
        self.command = command

    def _sync_execute_mcp_call(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """
        Dijalankan pada worker thread terpisah.
        Menghindari event loop utama terhenti akibat blocking pipe write/read.
        """
        import subprocess
        import json

        # ponytail: naive process execution; replace with persistent child process pool if latency-critical
        process = subprocess.Popen(
            self.command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )

        request_data = json.dumps(payload)
        stdout, stderr = process.communicate(input=request_data)

        if process.returncode != 0:
            raise RuntimeError(f"MCP worker failed: {stderr.strip()}")

        return json.loads(stdout)

    async def execute_tool(self, tool_name: str, arguments: dict, timeout_seconds: float = 30.0) -> Dict[str, Any]:
        """
        Async non-blocking wrapper dengan thread isolation dan strict timeout.
        """
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments}
        }

        try:
            # anyio.fail_after membatalkan context jika threshold terlewati
            with fail_after(timeout_seconds):
                result = await to_thread.run_sync(self._sync_execute_mcp_call, payload)
                return result
        except TimeoutError:
            raise TimeoutError(f"Tool '{tool_name}' timed out after {timeout_seconds} seconds")

async def main():
    client = DeadlockSafeMCPClient([sys.executable, "mcp_worker.py"])
    
    # Menjalankan concurrent calls dengan task group
    async with anyio.create_task_group() as tg:
        async def run_worker(tool: str, args: dict):
            try:
                res = await client.execute_tool(tool, args, timeout_seconds=10.0)
                print(f"Result from {tool}: {res}")
            except Exception as exc:
                print(f"Execution error on {tool}: {exc}", file=sys.stderr)

        tg.start_soon(run_worker, "fetch_data", {"source": "repo"})
        tg.start_soon(run_worker, "generate_diff", {"target": "feature"})

if __name__ == "__main__":
    anyio.run(main)

Trade-offs dan Pertimbangan Sistem

Meskipun isolasi menggunakan worker thread memecahkan issue event loop starvation, terdapat konsekuensi arsitektur yang perlu dipertimbangkan:

  • Overhead Konkurensi: Mengalihkan tugas ke worker thread via to_thread.run_sync menimbulkan overhead context switching. Jika agent Anda memanggil ribuan tools micro-latency, pertimbangkan model persistent async transport yang fully native non-blocking.
  • Child Process Memory Leak: Menjalankan short-lived child process memecahkan deadlock pipe buffer, tetapi berisiko meningkatkan footprint CPU jika proses sering di-spawn. Gunakan persistent socket atau named pipes via standard stream non-blocking jika throughput sangat tinggi.
  • Signal Handling: Pastikan process timeout selalu membersihkan subprocess yang terisolasi agar tidak menghasilkan zombie process pada host container Anda.