Module

_child

Persistent child process for MCP gateway communication.

1Class

Classes

ChildProcess 10 ▼
A persistent child process that speaks JSON-RPC on stdin/stdout. Thread-safe: a lock serializes al…

A persistent child process that speaks JSON-RPC on stdin/stdout.

Thread-safe: a lock serializes all calls to the same child. Auto-reconnects if the child process dies.

Methods

ensure_alive 0 ▼
Spawn or reconnect if the child process is dead.
def ensure_alive(self) -> None
send_call 2 dict[str, Any] ▼
Send a JSON-RPC request and return the result. Thread-safe.
def send_call(self, method: str, params: dict[str, Any]) -> dict[str, Any]
Parameters
Name Type Description
method —
params —
Returns
dict[str, Any]
fetch_tools 0 list[dict[str, Any]] ▼
Fetch tools/list from the child process.
def fetch_tools(self) -> list[dict[str, Any]]
Returns
list[dict[str, Any]]
is_idle 0 bool ▼
Check if the child has been idle longer than idle_timeout.
def is_idle(self) -> bool
Returns
bool
kill 0 ▼
Kill the child process.
def kill(self) -> None
Internal Methods 5 ▼
__init__ 3 ▼
def __init__(self, name: str, command: list[str], *, idle_timeout: float = 300.0) -> None
Parameters
Name Type Description
name —
command —
idle_timeout — Default:300.0
_spawn 0 ▼
Start the child process with persistent pipes.
def _spawn(self) -> None
_ensure_initialized 0 ▼
Send initialize if not already done.
def _ensure_initialized(self) -> None
_write_line 1 ▼
Write a line to the child's stdin.
def _write_line(self, line: str) -> None
Parameters
Name Type Description
line —
_read_line 0 str ▼
Read a line from the child's stdout with timeout. Uses a background thread so …
def _read_line(self) -> str

Read a line from the child's stdout with timeout.

Uses a background thread so we can enforce a deadline even when the child blocks without writing a newline.

Returns
str