INNER CODE UNIT · Python

_process_line

stevesolun/ctx · src/ctx/mcp_server/server.py:511

def _process_line(
    text: str,
    state: _ServerState,
    out_stream: BinaryIO,
) -> None:
    """Parse one JSON line, dispatch, emit response (or not, for notifications)."""
    started = time.perf_counter()
    try:
        frame = json.loads(text)
    except json.JSONDecodeError as exc:
        _write_error(
            out_stream,
            None,
            _ErrorCode.PARSE_ERROR,
            f"parse error: {exc}",
        )
        _record_mcp_request(
            payload=_safe_mcp_payload(

View source record →