Module

_child

Persistent child process for MCP gateway communication.

1Class

Classes

ChildProcess 12 ▼
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 3 dict[str, Any] ▼
Send a JSON-RPC request and return the result. Thread-safe. *timeout* override…
def send_call(self, method: str, params: dict[str, Any], *, timeout: float | None = None) -> dict[str, Any]

Send a JSON-RPC request and return the result. Thread-safe.

timeout overrides the instancerequest_timeoutfor this call.

Parameters
Name Type Description
method —
params —
timeout — Default:None
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 7 ▼
__init__ 4 ▼
def __init__(self, name: str, command: list[str], *, idle_timeout: float = 300.0, request_timeout: float = 30.0) -> None
Parameters
Name Type Description
name —
command —
idle_timeout — Default:300.0
request_timeout — Default:30.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
_graceful_kill 0 ▼
SIGTERM with grace period, then SIGKILL if needed. Must be called while holdin…
def _graceful_kill(self) -> None

SIGTERM with grace period, then SIGKILL if needed.

Must be called while holdingself._lock.

_write_line 1 ▼
Write a line to the child's stdin.
def _write_line(self, line: str) -> None
Parameters
Name Type Description
line —
_read_response 2 dict[str, Any] ▼
Read frames until the response matching *req_id* arrives.
def _read_response(self, req_id: int, *, timeout: float | None = None) -> dict[str, Any]
Parameters
Name Type Description
req_id —
timeout — Default:None
Returns
dict[str, Any]
_read_line 1 str ▼
Read a line from the child's stdout with timeout. Uses a background thread so …
def _read_line(self, *, timeout: float | None = None) -> 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.

Parameters
Name Type Description
timeout — Default:None
Returns
str